<!DOCTYPE html>



  


<html class="theme-next muse use-motion" lang="zh-Hans">
<head>
  <meta charset="UTF-8"/>
<meta http-equiv="X-UA-Compatible" content="IE=edge" />
<meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1"/>
<meta name="theme-color" content="#222">









<meta http-equiv="Cache-Control" content="no-transform" />
<meta http-equiv="Cache-Control" content="no-siteapp" />
















  
  
  <link href="/lib/fancybox/source/jquery.fancybox.css?v=2.1.5" rel="stylesheet" type="text/css" />







<link href="/lib/font-awesome/css/font-awesome.min.css?v=4.6.2" rel="stylesheet" type="text/css" />

<link href="/css/main.css?v=5.1.4" rel="stylesheet" type="text/css" />


  <link rel="apple-touch-icon" sizes="180x180" href="/images/apple-touch-icon-next.png?v=5.1.4">


  <link rel="icon" type="image/png" sizes="32x32" href="/images/favicon-32x32-next.png?v=5.1.4">


  <link rel="icon" type="image/png" sizes="16x16" href="/images/favicon-16x16-next.png?v=5.1.4">


  <link rel="mask-icon" href="/images/logo.svg?v=5.1.4" color="#222">





  <meta name="keywords" content="Hexo, NexT" />










<meta name="description" content="微信搜『中间件兴趣圈』，回复『Java』获取200本优质电子书">
<meta property="og:type" content="website">
<meta property="og:title" content="中间件兴趣圈">
<meta property="og:url" content="https://www.codingw.net/page/11/index.html">
<meta property="og:site_name" content="中间件兴趣圈">
<meta property="og:description" content="微信搜『中间件兴趣圈』，回复『Java』获取200本优质电子书">
<meta property="og:locale">
<meta property="article:author" content="中间件兴趣圈">
<meta property="article:tag" content="中间件">
<meta name="twitter:card" content="summary">



<script type="text/javascript" id="hexo.configurations">
  var NexT = window.NexT || {};
  var CONFIG = {
    root: '',
    scheme: 'Muse',
    version: '5.1.4',
    sidebar: {"position":"left","display":"post","offset":12,"b2t":false,"scrollpercent":false,"onmobile":false},
    fancybox: true,
    tabs: true,
    motion: {"enable":true,"async":false,"transition":{"post_block":"fadeIn","post_header":"slideDownIn","post_body":"slideDownIn","coll_header":"slideLeftIn","sidebar":"slideUpIn"}},
    duoshuo: {
      userId: '0',
      author: '博主'
    },
    algolia: {
      applicationID: '',
      apiKey: '',
      indexName: '',
      hits: {"per_page":10},
      labels: {"input_placeholder":"Search for Posts","hits_empty":"We didn't find any results for the search: ${query}","hits_stats":"${hits} results found in ${time} ms"}
    }
  };
</script>



  <link rel="canonical" href="https://www.codingw.net/page/11/"/>





  <title>中间件兴趣圈</title>
  








<meta name="generator" content="Hexo 5.4.0"></head>

<body itemscope itemtype="http://schema.org/WebPage" lang="zh-Hans">

  
  
    
  

  <div class="container sidebar-position-left 
  page-home">
    <div class="headband"></div>

    <header id="header" class="header" itemscope itemtype="http://schema.org/WPHeader">
      <div class="header-inner"><div class="site-brand-wrapper">
  <div class="site-meta ">
    

    <div class="custom-logo-site-title">
      <a href="/"  class="brand" rel="start">
        <span class="logo-line-before"><i></i></span>
        <span class="site-title">中间件兴趣圈</span>
        <span class="logo-line-after"><i></i></span>
      </a>
    </div>
      
        <p class="site-subtitle">微信搜『中间件兴趣圈』，回复『Java』获取200本优质电子书</p>
      
  </div>

  <div class="site-nav-toggle">
    <button>
      <span class="btn-bar"></span>
      <span class="btn-bar"></span>
      <span class="btn-bar"></span>
    </button>
  </div>
</div>

<nav class="site-nav">
  

  
    <ul id="menu" class="menu">
      
        
        <li class="menu-item menu-item-home">
          <a href="/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-home"></i> <br />
            
            首页
          </a>
        </li>
      
        
        <li class="menu-item menu-item-categories">
          <a href="/categories/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-question-circle"></i> <br />
            
            分类
          </a>
        </li>
      
        
        <li class="menu-item menu-item-archives">
          <a href="/archives/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-question-circle"></i> <br />
            
            归档
          </a>
        </li>
      

      
    </ul>
  

  
</nav>



 </div>
    </header>

    <main id="main" class="main">
      <div class="main-inner">
        <div class="content-wrap">
          <div id="content" class="content">
            
  <section id="posts" class="posts-expand">
    
      

  

  
  
  

  <article class="post post-type-normal" itemscope itemtype="http://schema.org/Article">
  
  
  
  <div class="post-block">
    <link itemprop="mainEntityOfPage" href="https://www.codingw.net/posts/b9e3e3ee.html">

    <span hidden itemprop="author" itemscope itemtype="http://schema.org/Person">
      <meta itemprop="name" content="">
      <meta itemprop="description" content="">
      <meta itemprop="image" content="/images/avatar.gif">
    </span>

    <span hidden itemprop="publisher" itemscope itemtype="http://schema.org/Organization">
      <meta itemprop="name" content="中间件兴趣圈">
    </span>

    
      <header class="post-header">

        
        
          <h1 class="post-title" itemprop="name headline">
                
                <a class="post-title-link" href="/posts/b9e3e3ee.html" itemprop="url">Kafka与RocketMQ性能对比大揭秘</a></h1>
        

        <div class="post-meta">
          <span class="post-time">
            
              <span class="post-meta-item-icon">
                <i class="fa fa-calendar-o"></i>
              </span>
              
                <span class="post-meta-item-text">发表于</span>
              
              <time title="创建于" itemprop="dateCreated datePublished" datetime="2020-10-01T22:37:01+08:00">
                2020-10-01
              </time>
            

            

            
          </span>

          
            <span class="post-category" >
            
              <span class="post-meta-divider">|</span>
            
              <span class="post-meta-item-icon">
                <i class="fa fa-folder-o"></i>
              </span>
              
                <span class="post-meta-item-text">分类于</span>
              
              
                <span itemprop="about" itemscope itemtype="http://schema.org/Thing">
                  <a href="/categories/kafka/" itemprop="url" rel="index">
                    <span itemprop="name">kafka</span>
                  </a>
                </span>

                
                
              
            </span>
          

          
            
          

          
          
             <span id="/posts/b9e3e3ee.html" class="leancloud_visitors" data-flag-title="Kafka与RocketMQ性能对比大揭秘">
               <span class="post-meta-divider">|</span>
               <span class="post-meta-item-icon">
                 <i class="fa fa-eye"></i>
               </span>
               
                 <span class="post-meta-item-text">阅读次数&#58;</span>
               
                 <span class="leancloud-visitors-count"></span>
             </span>
          

          

          

          

        </div>
      </header>
    

    
    
    
    <div class="post-body" itemprop="articleBody">

      
      

      
        
          <p>在双十一过程中投入同样的硬件资源，Kafka 搭建的日志集群单个Topic可以达到几百万的TPS，而使用RocketMQ组件的核心业务集群，集群TPS只能达到几十万TPS，这样的现象激发了我对两者性能方面的思考。</p>
<blockquote>
<p>温馨提示：TPS只是众多性能指标中的一个，我们在做技术选型方面要从多方面考虑，本文并不打算就消息中间件选型方面投入太多笔墨，重点想尝试剖析两者在性能方面的设计思想。</p>
</blockquote>
          <!--noindex-->
          <div class="post-button text-center">
            <a class="btn" href="/posts/b9e3e3ee.html#more" rel="contents">
              阅读全文 &raquo;
            </a>
          </div>
          <!--/noindex-->
        
      
    </div>
    
    
    

    

    

    

    <footer class="post-footer">
      

      

      

      
      
        <div class="post-eof"></div>
      
    </footer>
  </div>
  
  
  
  </article>


    
      

  

  
  
  

  <article class="post post-type-normal" itemscope itemtype="http://schema.org/Article">
  
  
  
  <div class="post-block">
    <link itemprop="mainEntityOfPage" href="https://www.codingw.net/posts/f03d9942.html">

    <span hidden itemprop="author" itemscope itemtype="http://schema.org/Person">
      <meta itemprop="name" content="">
      <meta itemprop="description" content="">
      <meta itemprop="image" content="/images/avatar.gif">
    </span>

    <span hidden itemprop="publisher" itemscope itemtype="http://schema.org/Organization">
      <meta itemprop="name" content="中间件兴趣圈">
    </span>

    
      <header class="post-header">

        
        
          <h1 class="post-title" itemprop="name headline">
                
                <a class="post-title-link" href="/posts/f03d9942.html" itemprop="url">源码分析 RocketMQ DLedger 多副本存储实现</a></h1>
        

        <div class="post-meta">
          <span class="post-time">
            
              <span class="post-meta-item-icon">
                <i class="fa fa-calendar-o"></i>
              </span>
              
                <span class="post-meta-item-text">发表于</span>
              
              <time title="创建于" itemprop="dateCreated datePublished" datetime="2020-09-30T23:47:01+08:00">
                2020-09-30
              </time>
            

            

            
          </span>

          
            <span class="post-category" >
            
              <span class="post-meta-divider">|</span>
            
              <span class="post-meta-item-icon">
                <i class="fa fa-folder-o"></i>
              </span>
              
                <span class="post-meta-item-text">分类于</span>
              
              
                <span itemprop="about" itemscope itemtype="http://schema.org/Thing">
                  <a href="/categories/rocketmq/" itemprop="url" rel="index">
                    <span itemprop="name">rocketmq</span>
                  </a>
                </span>

                
                
              
            </span>
          

          
            
          

          
          
             <span id="/posts/f03d9942.html" class="leancloud_visitors" data-flag-title="源码分析 RocketMQ DLedger 多副本存储实现">
               <span class="post-meta-divider">|</span>
               <span class="post-meta-item-icon">
                 <i class="fa fa-eye"></i>
               </span>
               
                 <span class="post-meta-item-text">阅读次数&#58;</span>
               
                 <span class="leancloud-visitors-count"></span>
             </span>
          

          

          

          

        </div>
      </header>
    

    
    
    
    <div class="post-body" itemprop="articleBody">

      
      

      
        
          <p>RocketMQ DLedger 的存储实现思路与 RocketMQ 的存储实现思路相似，本文就不再从源码角度详细剖析其实现，只是点出其实现关键点。我们不妨简单回顾一下 CommitLog 文件、ConsumeQueue 文件设计思想。</p>
<p>其文件组成形式如下：<br><img src="https://img-blog.csdnimg.cn/20190831215052708.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70" alt="在这里插入图片描述"><br>正如上图所示，多个 commitlog 文件组成一个逻辑上的连续文件，使用 MappedFileQueue 表示，单个 commitlog 文件使用 MappedFile 表示。</p>
<blockquote>
<p>温馨提示：如果想详细了解 RocketMQ 关于存储部分的讲解，可以关注笔者的《RocketMQ 技术内幕》一书。</p>
</blockquote>
<h2 id="1、DLedger-存储相关类图"><a href="#1、DLedger-存储相关类图" class="headerlink" title="1、DLedger 存储相关类图"></a>1、DLedger 存储相关类图</h2><p><img src="https://img-blog.csdnimg.cn/20190831215151796.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70" alt="在这里插入图片描述"></p>
<h3 id="1-1-DLedgerStore"><a href="#1-1-DLedgerStore" class="headerlink" title="1.1 DLedgerStore"></a>1.1 DLedgerStore</h3><p>存储抽象类，定义如下核心方法：</p>
<ul>
<li>public abstract DLedgerEntry appendAsLeader(DLedgerEntry entry)<br>向主节点追加日志(数据)。</li>
<li>public abstract DLedgerEntry appendAsFollower(DLedgerEntry entry, long leaderTerm, String leaderId)<br>向从节点同步日志。</li>
<li>public abstract DLedgerEntry get(Long index)<br>根据日志下标查找日志。</li>
<li>public abstract long getCommittedIndex()<br>获取已提交的下标。</li>
<li>public abstract long getLedgerEndTerm()<br>获取 Leader 当前最大的投票轮次。</li>
<li>public abstract long getLedgerEndIndex()<br>获取 Leader 下一条日志写入的下标（最新日志的下标）。</li>
<li>public abstract long getLedgerBeginIndex()<br>获取 Leader 第一条消息的下标。</li>
<li>public void updateCommittedIndex(long term, long committedIndex)<br>更新commitedIndex的值，为空实现，由具体的存储子类实现。</li>
<li>protected void updateLedgerEndIndexAndTerm()<br>更新 Leader 维护的 ledgerEndIndex 和 ledgerEndTerm 。</li>
<li>public void flush()<br>刷写，空方法，由具体子类实现。</li>
<li>public long truncate(DLedgerEntry entry, long leaderTerm, String leaderId)<br>删除日志，空方法，由具体子类实现。</li>
<li>public void startup()<br>启动存储管理器，空方法，由具体子类实现。</li>
<li>public void shutdown()<br>关闭存储管理器，空方法，由具体子类实现。</li>
</ul>
<h3 id="1-2-DLedgerMemoryStore"><a href="#1-2-DLedgerMemoryStore" class="headerlink" title="1.2 DLedgerMemoryStore"></a>1.2 DLedgerMemoryStore</h3><p>Dledger 基于内存实现的日志存储。</p>
<h3 id="1-3-DLedgerMmapFileStore"><a href="#1-3-DLedgerMmapFileStore" class="headerlink" title="1.3  DLedgerMmapFileStore"></a>1.3  DLedgerMmapFileStore</h3><p>基于文件内存映射机制的存储实现。其核心属性如下：</p>
<ul>
<li>long ledgerBeginIndex =  -1<br>日志的起始索引，默认为 -1。<br>l- ong ledgerEndIndex = -1<br>下一条日志下标，默认为 -1。</li>
<li>long committedIndex = -1<br>已提交的日志索引。</li>
<li>long ledgerEndTerm<br>当前最大的投票轮次。</li>
<li>DLedgerConfig dLedgerConfig<br>DLedger 的配置信息。</li>
<li>MemberState memberState<br>状态机。</li>
<li>MmapFileList dataFileList<br>日志文件(数据文件)的内存映射Queue。</li>
<li>MmapFileList indexFileList<br>索引文件的内存映射文件集合。（可对标 RocketMQ MappedFIleQueue )。</li>
<li>ThreadLocal&lt; ByteBuffer&gt; localIndexBuffer<br>本地线程变量，用来缓存索引ByteBuffer。</li>
<li>ThreadLocal&lt; ByteBuffer&gt; localEntryBuffer<br>本地线程变量，用来缓存数据索引ByteBuffer。</li>
<li>FlushDataService flushDataService<br>数据文件刷盘线程。</li>
<li>CleanSpaceService cleanSpaceService<br>清除过期日志文件线程。</li>
<li>boolean isDiskFull = false<br>磁盘是否已满。</li>
<li>long lastCheckPointTimeMs<br>上一次检测点（时间戳）。</li>
<li>AtomicBoolean hasLoaded<br>是否已经加载，主要用来避免重复加载(初始化)日志文件。</li>
<li>AtomicBoolean hasRecovered<br> ​    是否已恢复。</li>
</ul>
          <!--noindex-->
          <div class="post-button text-center">
            <a class="btn" href="/posts/f03d9942.html#more" rel="contents">
              阅读全文 &raquo;
            </a>
          </div>
          <!--/noindex-->
        
      
    </div>
    
    
    

    

    

    

    <footer class="post-footer">
      

      

      

      
      
        <div class="post-eof"></div>
      
    </footer>
  </div>
  
  
  
  </article>


    
      

  

  
  
  

  <article class="post post-type-normal" itemscope itemtype="http://schema.org/Article">
  
  
  
  <div class="post-block">
    <link itemprop="mainEntityOfPage" href="https://www.codingw.net/posts/57908f75.html">

    <span hidden itemprop="author" itemscope itemtype="http://schema.org/Person">
      <meta itemprop="name" content="">
      <meta itemprop="description" content="">
      <meta itemprop="image" content="/images/avatar.gif">
    </span>

    <span hidden itemprop="publisher" itemscope itemtype="http://schema.org/Organization">
      <meta itemprop="name" content="中间件兴趣圈">
    </span>

    
      <header class="post-header">

        
        
          <h1 class="post-title" itemprop="name headline">
                
                <a class="post-title-link" href="/posts/57908f75.html" itemprop="url">基于 Flink 实现解决数据库分库分表任务拆分</a></h1>
        

        <div class="post-meta">
          <span class="post-time">
            
              <span class="post-meta-item-icon">
                <i class="fa fa-calendar-o"></i>
              </span>
              
                <span class="post-meta-item-text">发表于</span>
              
              <time title="创建于" itemprop="dateCreated datePublished" datetime="2020-09-29T22:32:35+08:00">
                2020-09-29
              </time>
            

            

            
          </span>

          
            <span class="post-category" >
            
              <span class="post-meta-divider">|</span>
            
              <span class="post-meta-item-icon">
                <i class="fa fa-folder-o"></i>
              </span>
              
                <span class="post-meta-item-text">分类于</span>
              
              
                <span itemprop="about" itemscope itemtype="http://schema.org/Thing">
                  <a href="/categories/flink/" itemprop="url" rel="index">
                    <span itemprop="name">flink</span>
                  </a>
                </span>

                
                
              
            </span>
          

          
            
          

          
          
             <span id="/posts/57908f75.html" class="leancloud_visitors" data-flag-title="基于 Flink 实现解决数据库分库分表任务拆分">
               <span class="post-meta-divider">|</span>
               <span class="post-meta-item-icon">
                 <i class="fa fa-eye"></i>
               </span>
               
                 <span class="post-meta-item-text">阅读次数&#58;</span>
               
                 <span class="leancloud-visitors-count"></span>
             </span>
          

          

          

          

        </div>
      </header>
    

    
    
    
    <div class="post-body" itemprop="articleBody">

      
      

      
        
          <h2 id="1、场景描述"><a href="#1、场景描述" class="headerlink" title="1、场景描述"></a>1、场景描述</h2><p>例如订单库进行了分库分表，其示例如下图所示：<br><img src="https://img-blog.csdnimg.cn/20201115203825639.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述"><br>现在的需求是希望创建一个任务就将数据同步到MQ集群，而不是为每一个数据库实例单独创建一个任务，将其数据导入到MQ集群，因为同步任务除了库不同之外，表的结构、数据映射规则都是一致的。</p>
<h2 id="2、flinkx-的解决方案详解"><a href="#2、flinkx-的解决方案详解" class="headerlink" title="2、flinkx 的解决方案详解"></a>2、flinkx 的解决方案详解</h2><h4 id="2-1-fink-Stream-API-开发基本流程"><a href="#2-1-fink-Stream-API-开发基本流程" class="headerlink" title="2.1 fink Stream API 开发基本流程"></a>2.1 fink Stream API 开发基本流程</h4><p>使用 Flink Stream API 编程的通用步骤如下图所示：<br><img src="https://img-blog.csdnimg.cn/20201115203839996.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述"></p>
<blockquote>
<p>温馨提示：有关 Stream API 的详细内容将在后续的文章中展开，本文主要是关注 InputFormatSourceFunction，重点关注数据源的拆分。</p>
</blockquote>
<h4 id="2-2-flinkx-Reader-数据源-核心类图"><a href="#2-2-flinkx-Reader-数据源-核心类图" class="headerlink" title="2.2 flinkx Reader(数据源)核心类图"></a>2.2 flinkx Reader(数据源)核心类图</h4><p><img src="https://img-blog.csdnimg.cn/20201115203855115.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述"><br>在 flinkx 中将不同的数据源封装成一个个 Reader，其基类为 BaseDataReader，上图中主要罗列了如下几个关键的类体系：</p>
<ul>
<li><p>InputFormat<br>flink 核心API，主要是对输入源进行数据切分、读取数据的抽象，其核心接口说明如下：</p>
<ul>
<li><p>void configure(Configuration parameters)<br>对输入源进行额外的配置，该方法在 Input 的生命周期中只需调用一次。</p>
</li>
<li><p>BaseStatistics getStatistics(BaseStatistics cachedStatistics)<br>返回 input 的统计数据，如果不需要统计，在实现的时候可以直接返回 null。</p>
</li>
<li><p>T[] createInputSplits(int minNumSplits)<br>对输入数据进行数据切片，使之支持并行处理，数据切片相关类体系见：InputSplit。</p>
</li>
<li><p>InputSplitAssigner getInputSplitAssigner(T[] inputSplits)<br>获取 InputSplit 分配器，主要是在具体执行任务时如何获取下一个 InputSplit，其声明如下图所示：<br><img src="https://img-blog.csdnimg.cn/20201115204001187.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_right#pic_center" alt="在这里插入图片描述"></p>
</li>
<li><p>void open(T split)<br>根据指定的数据分片 (InputSplit) 打开数据通道。为了加深对该方法的理解，下面看一下 Flinkx 关于 jdbc、es 的写入示例：<br><img src="https://img-blog.csdnimg.cn/20201115203933977.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_right#pic_center" alt="在这里插入图片描述"></p>
</li>
</ul>
</li>
</ul>
<ul>
<li><p>boolean reachedEnd()<br>数据是否已结束，在 Flink 中通常 InputFormat 的数据源通常表示有界数据 (DataSet)。</p>
</li>
<li><p>OT nextRecord(OT reuse)</p>
<p>从通道中获取下一条记录。</p>
</li>
<li><p>void close()<br>关闭。</p>
</li>
<li><p>InputSplit<br>数据分片根接口，只定义了如下方法：</p>
<ul>
<li>int getSplitNumber()<br>获取当前分片所在所有分片中的序号。</li>
</ul>
<p>本文先简单介绍一下其通用实现子类：GenericInputSplit。</p>
<ul>
<li>int partitionNumber<br>当前 split 所在的序号</li>
<li>int totalNumberOfPartitions<br>总分片数</li>
</ul>
<p>为了方便理解我们可以思考一下如下场景，对于一个数据量超过千万级别的表，在进行数据切分时可以考虑使用10个线程，即切割成 10分，那每一个数据线程查询数据时可以 id % totalNumberOfPartitions = partitionNumber，进行数据读取。</p>
</li>
<li><p>SourceFunction<br>Flink 源的抽象定义。</p>
<ul>
<li><p>RichFunction<br>富函数，定义了生命周期、可获取运行时环境上下文。</p>
</li>
<li><p>ParallelSourceFunction<br>支持并行的 source function。</p>
</li>
<li><p>RichParallelSourceFunction</p>
<p>并行的富函数</p>
</li>
<li><p>InputFormatSourceFunction</p>
<p>Flink 默认提供的 RichParallelSourceFunction 实现类，可以当成是RichParallelSourceFunction 的通用写法，其内部的数据读取逻辑由 InputFormat 实现。</p>
</li>
</ul>
</li>
<li><p>BaseDataReader</p>
<p>flinkx 数据读取基类，在 flinkx 中将所有的数据读取源封装成 Reader 。</p>
</li>
</ul>
<h4 id="2-3-flinkx-Reader构建-DataStream-流程"><a href="#2-3-flinkx-Reader构建-DataStream-流程" class="headerlink" title="2.3 flinkx Reader构建 DataStream 流程"></a>2.3 flinkx Reader构建 DataStream 流程</h4><p>经过了上面类图的梳理，大家应该 flink 中提到的上述类的含义有了一个大概的理解，但如何运用呢？接下来将通过查阅 flinkx 的 DistributedJdbcDataReader(BaseDataReader的子类)的 readData 调用流程，体会一下其使用方法。<br><img src="https://img-blog.csdnimg.cn/20201115204021519.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述"><br>基本遵循创建 InputFormat、从而创建对应的 SourceFunction，然后通过 StreamExecutionEnvironment 的 addSource 方法将 SourceFunction 创建对应的 DataStreamSource。</p>
          <!--noindex-->
          <div class="post-button text-center">
            <a class="btn" href="/posts/57908f75.html#more" rel="contents">
              阅读全文 &raquo;
            </a>
          </div>
          <!--/noindex-->
        
      
    </div>
    
    
    

    

    

    

    <footer class="post-footer">
      

      

      

      
      
        <div class="post-eof"></div>
      
    </footer>
  </div>
  
  
  
  </article>


    
      

  

  
  
  

  <article class="post post-type-normal" itemscope itemtype="http://schema.org/Article">
  
  
  
  <div class="post-block">
    <link itemprop="mainEntityOfPage" href="https://www.codingw.net/posts/12736650.html">

    <span hidden itemprop="author" itemscope itemtype="http://schema.org/Person">
      <meta itemprop="name" content="">
      <meta itemprop="description" content="">
      <meta itemprop="image" content="/images/avatar.gif">
    </span>

    <span hidden itemprop="publisher" itemscope itemtype="http://schema.org/Organization">
      <meta itemprop="name" content="中间件兴趣圈">
    </span>

    
      <header class="post-header">

        
        
          <h1 class="post-title" itemprop="name headline">
                
                <a class="post-title-link" href="/posts/12736650.html" itemprop="url">源码阅读技巧篇：RocketMQ DLedger 多副本即主从切换专栏回顾</a></h1>
        

        <div class="post-meta">
          <span class="post-time">
            
              <span class="post-meta-item-icon">
                <i class="fa fa-calendar-o"></i>
              </span>
              
                <span class="post-meta-item-text">发表于</span>
              
              <time title="创建于" itemprop="dateCreated datePublished" datetime="2020-09-28T21:55:35+08:00">
                2020-09-28
              </time>
            

            

            
          </span>

          
            <span class="post-category" >
            
              <span class="post-meta-divider">|</span>
            
              <span class="post-meta-item-icon">
                <i class="fa fa-folder-o"></i>
              </span>
              
                <span class="post-meta-item-text">分类于</span>
              
              
                <span itemprop="about" itemscope itemtype="http://schema.org/Thing">
                  <a href="/categories/rocketmq/" itemprop="url" rel="index">
                    <span itemprop="name">rocketmq</span>
                  </a>
                </span>

                
                
              
            </span>
          

          
            
          

          
          
             <span id="/posts/12736650.html" class="leancloud_visitors" data-flag-title="源码阅读技巧篇：RocketMQ DLedger 多副本即主从切换专栏回顾">
               <span class="post-meta-divider">|</span>
               <span class="post-meta-item-icon">
                 <i class="fa fa-eye"></i>
               </span>
               
                 <span class="post-meta-item-text">阅读次数&#58;</span>
               
                 <span class="leancloud-visitors-count"></span>
             </span>
          

          

          

          

        </div>
      </header>
    

    
    
    
    <div class="post-body" itemprop="articleBody">

      
      

      
        
          <p>RocketMQ DLedger 多副本即主从切换专栏总共包含9篇文章，时间跨度大概为2个月的时间，笔者觉得授人以鱼不如授人以渔，借以这个系列来展示该系列的创作始末，展示笔者阅读源码的技巧。</p>
<p>首先在下决心研读  RocketMQ DLedger 多副本(主从切换)的源码之前，首先还是要通过官方的分享、百度等途径对该功能进行一些基本的了解。</p>
<p>我们了解到 RocketMQ 在 4.5.0 之前提供了主从同步功能，即当主节点宕机后，消费端可以继续从从节点上消费消息，但无法继续向该复制组发送消息。RocketMQ 4.5.0版本引入了多副本机制，即 DLedger，支持主从切换，即当一个复制组内的主节点宕机后，会在该复制组内触发重新选主，选主完成后即可继续提供消息写功能。同时还了解到 rocketmq 主从切换是基于 raft 协议的。</p>
<p>raft 协议是何许人也，我猜想大部分读者对这个名词并不陌生，但像笔者一样只是听过其大体作用但并未详细学习的应该也不在少数，故我觉得看 RocketMQ DLedger 多副本即主从切换之前应该重点了解 raft 协议。</p>
<p>1、<a target="_blank" rel="noopener" href="https://blog.csdn.net/prestigeding/article/details/99101912">RocketMQ 多副本前置篇：初探raft协议</a></p>
<p>本文主要根据 raft 官方提供的动画来学习了解 raft 协议，从本文基本得知了 raft 协议主要包含两个重要部分：选主 以及 日志复制。在了解了 raft 协议的选主、日志复制的基本实现后，然后就可以步入到 RocketMQ DLedger 多副本即主从切换的源码研究了，以探究大神是如何实现 raft 协议的。同时在了解到了 raft 协议的选主部分内容后，自己也可以简单的思考，如果自己去实现 raft 协议，应该要实现哪些关键点，当时我的思考如下：<br><img src="https://img-blog.csdnimg.cn/20191020212502611.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70" alt="在这里插入图片描述"><br>这样在看源码时更加有针对性，不至于在阅读源码过程中“迷失”。</p>
<p>2、<a target="_blank" rel="noopener" href="https://blog.csdn.net/prestigeding/article/details/99697323">源码分析 RocketMQ DLedger 多副本之 Leader 选主</a></p>
<p>本文按照上一篇的思路，重点对 DLedgerLeaderElector 的实现进行了详细分析，特别是其内部的状态机流转，最后也给出一张流程图对选主过程进行一个简单的梳理与总结。</p>
          <!--noindex-->
          <div class="post-button text-center">
            <a class="btn" href="/posts/12736650.html#more" rel="contents">
              阅读全文 &raquo;
            </a>
          </div>
          <!--/noindex-->
        
      
    </div>
    
    
    

    

    

    

    <footer class="post-footer">
      

      

      

      
      
        <div class="post-eof"></div>
      
    </footer>
  </div>
  
  
  
  </article>


    
      

  

  
  
  

  <article class="post post-type-normal" itemscope itemtype="http://schema.org/Article">
  
  
  
  <div class="post-block">
    <link itemprop="mainEntityOfPage" href="https://www.codingw.net/posts/550fbf45.html">

    <span hidden itemprop="author" itemscope itemtype="http://schema.org/Person">
      <meta itemprop="name" content="">
      <meta itemprop="description" content="">
      <meta itemprop="image" content="/images/avatar.gif">
    </span>

    <span hidden itemprop="publisher" itemscope itemtype="http://schema.org/Organization">
      <meta itemprop="name" content="中间件兴趣圈">
    </span>

    
      <header class="post-header">

        
        
          <h1 class="post-title" itemprop="name headline">
                
                <a class="post-title-link" href="/posts/550fbf45.html" itemprop="url">寻找一把进入 Alibaba Sentinel 的钥匙(文末附流程图)</a></h1>
        

        <div class="post-meta">
          <span class="post-time">
            
              <span class="post-meta-item-icon">
                <i class="fa fa-calendar-o"></i>
              </span>
              
                <span class="post-meta-item-text">发表于</span>
              
              <time title="创建于" itemprop="dateCreated datePublished" datetime="2020-09-27T22:43:35+08:00">
                2020-09-27
              </time>
            

            

            
          </span>

          
            <span class="post-category" >
            
              <span class="post-meta-divider">|</span>
            
              <span class="post-meta-item-icon">
                <i class="fa fa-folder-o"></i>
              </span>
              
                <span class="post-meta-item-text">分类于</span>
              
              
                <span itemprop="about" itemscope itemtype="http://schema.org/Thing">
                  <a href="/categories/sentinel/" itemprop="url" rel="index">
                    <span itemprop="name">sentinel</span>
                  </a>
                </span>

                
                
              
            </span>
          

          
            
          

          
          
             <span id="/posts/550fbf45.html" class="leancloud_visitors" data-flag-title="寻找一把进入 Alibaba Sentinel 的钥匙(文末附流程图)">
               <span class="post-meta-divider">|</span>
               <span class="post-meta-item-icon">
                 <i class="fa fa-eye"></i>
               </span>
               
                 <span class="post-meta-item-text">阅读次数&#58;</span>
               
                 <span class="leancloud-visitors-count"></span>
             </span>
          

          

          

          

        </div>
      </header>
    

    
    
    
    <div class="post-body" itemprop="articleBody">

      
      

      
        
          <p>经过前面几篇文章的铺垫，我们正式来探讨 Sentinel 的 entry 方法的实现流程。</p>
<p>即探究进入 Alibaba Sentinel 核心的一把钥匙。<br>无论是从 Sentinel 适配 Dubbo 也好，还是 SphU 源码中的注释中能看出，对一个资源进行限流或熔断，通常需要调用 SphU 的 entry 方法，例如如下示例代码。</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">foo</span><span class="params">()</span> </span>&#123;</span><br><span class="line">	Entry entry = <span class="keyword">null</span>;</span><br><span class="line">	<span class="keyword">try</span> &#123;</span><br><span class="line">		entry = SphU.entry(<span class="string">&quot;abc&quot;</span>);</span><br><span class="line">	&#125; <span class="keyword">catch</span> (BlockException blockException) &#123;</span><br><span class="line">		<span class="comment">// when goes there, it is blocked</span></span><br><span class="line">		<span class="comment">// add blocked handle logic here</span></span><br><span class="line">	&#125; <span class="keyword">catch</span> (Throwable bizException) &#123;</span><br><span class="line">		<span class="comment">// business exception</span></span><br><span class="line">		Tracer.trace(bizException);</span><br><span class="line">	&#125; <span class="keyword">finally</span> &#123;</span><br><span class="line">		<span class="comment">// ensure finally be executed</span></span><br><span class="line">		<span class="keyword">if</span> (entry != <span class="keyword">null</span>)&#123;</span><br><span class="line">			entry.exit();</span><br><span class="line">		&#125;</span><br><span class="line">	&#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>那本文将来探讨 SphU.entry 的实现原理。SphU 类定义了很多 entry 重载方法，我们就以下面这个方法为例来探究其实现原理。</p>
<h2 id="1、SphU-entry-流程分析"><a href="#1、SphU-entry-流程分析" class="headerlink" title="1、SphU.entry 流程分析"></a>1、SphU.entry 流程分析</h2><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">static</span> Entry <span class="title">entry</span><span class="params">(String name, EntryType type, <span class="keyword">int</span> count, Object... args)</span> <span class="keyword">throws</span>  BlockException </span>&#123;  <span class="comment">// @1</span></span><br><span class="line">	<span class="keyword">return</span> Env.sph.entry(name, type, count, args);  <span class="comment">// @2</span></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>代码@1：我们先来简单介绍其核心参数的含义：</p>
<ul>
<li>String name<br>资源的名称。</li>
<li>EntryType type<br>进入资源的方式，主要包含 EntryType.IN、EntryType.OUT。</li>
<li>int count<br>可以理解为本次进入需要消耗的“令牌数”。</li>
<li>Object… args<br>其他参数。</li>
</ul>
<p>代码@2：调用 Env.sph.entry 的方法，其最终会调用 CtSph 的 entry 方法。</p>
<p>接下来我们将重点查看 CtSph 的 entry 方法。</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> Entry <span class="title">entry</span><span class="params">(String name, EntryType type, <span class="keyword">int</span> count, Object... args)</span> <span class="keyword">throws</span> BlockException </span>&#123;</span><br><span class="line">    StringResourceWrapper resource = <span class="keyword">new</span> StringResourceWrapper(name, type); <span class="comment">// @1</span></span><br><span class="line">    <span class="keyword">return</span> entry(resource, count, args);  <span class="comment">// @2</span></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>代码@1：由于该方法用来表示资源的方式为一个字符串，故创建一个 StringResourceWrapper  对象来表示一个 Sentinel 中的资源，另外一个实现为 MethodResourceWrapper，用来表示方法类的资源。</p>
<p>代码@2：继续调用 CtSph 的另外一个 entry 重载方法，最终会调用 entryWithPriority 方法。</p>
<p>CtSph#entryWithPriority </p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">private</span> Entry <span class="title">entryWithPriority</span><span class="params">(ResourceWrapper resourceWrapper, <span class="keyword">int</span> count, <span class="keyword">boolean</span> prioritized, Object... args)</span> <span class="comment">// @1</span></span></span><br><span class="line"><span class="function">        <span class="keyword">throws</span> BlockException </span>&#123;</span><br><span class="line">    Context context = ContextUtil.getContext();  <span class="comment">// @2</span></span><br><span class="line">    <span class="keyword">if</span> (context <span class="keyword">instanceof</span> NullContext) &#123;</span><br><span class="line">      <span class="keyword">return</span> <span class="keyword">new</span> CtEntry(resourceWrapper, <span class="keyword">null</span>, context); </span><br><span class="line">    &#125;</span><br><span class="line">    <span class="keyword">if</span> (context == <span class="keyword">null</span>) &#123;</span><br><span class="line">        <span class="comment">// Using default context.</span></span><br><span class="line">        context = InternalContextUtil.internalEnter(Constants.CONTEXT_DEFAULT_NAME);</span><br><span class="line">    &#125;</span><br><span class="line">   <span class="keyword">if</span> (!Constants.ON) &#123;   <span class="comment">// @3</span></span><br><span class="line">        <span class="keyword">return</span> <span class="keyword">new</span> CtEntry(resourceWrapper, <span class="keyword">null</span>, context);</span><br><span class="line">    &#125;</span><br><span class="line">   </span><br><span class="line">    ProcessorSlot&lt;Object&gt; chain = lookProcessChain(resourceWrapper);   <span class="comment">// @4</span></span><br><span class="line">    <span class="keyword">if</span> (chain == <span class="keyword">null</span>) &#123;</span><br><span class="line">	    <span class="keyword">return</span> <span class="keyword">new</span> CtEntry(resourceWrapper, <span class="keyword">null</span>, context);</span><br><span class="line">    &#125;</span><br><span class="line">    Entry e = <span class="keyword">new</span> CtEntry(resourceWrapper, chain, context);     <span class="comment">// @5</span></span><br><span class="line">    <span class="keyword">try</span> &#123;</span><br><span class="line">	    chain.entry(context, resourceWrapper, <span class="keyword">null</span>, count, prioritized, args);   <span class="comment">// @6</span></span><br><span class="line">    &#125; <span class="keyword">catch</span> (BlockException e1) &#123;                                                                    <span class="comment">// @7</span></span><br><span class="line">	    e.exit(count, args);</span><br><span class="line">        <span class="keyword">throw</span> e1;</span><br><span class="line">    &#125; <span class="keyword">catch</span> (Throwable e1) &#123;</span><br><span class="line">        RecordLog.info(<span class="string">&quot;Sentinel unexpected exception&quot;</span>, e1);</span><br><span class="line">    &#125;</span><br><span class="line">    <span class="keyword">return</span> e;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>代码@1：我们先来介绍一下该方法的参数：</p>
<ul>
<li>ResourceWrapper resourceWrapper<br>资源的包装类型，可以是字符串类型的资源描述，也可以是方法类的。</li>
<li>int count<br>此次需要消耗的令牌。</li>
<li>boolean prioritized<br>是否注重优先级。</li>
<li>Object… args<br>额外参数。</li>
</ul>
<p>代码@2：获取方法调用的上下文环境，上下环境对象存储在线程本地变量：ThreadLocal 中，这里先“剧透”一下，上下文环境中存储的是整个调用链，后续文章会重点介绍。</p>
<p>代码@3：Sentinel 提供一个全局关闭的开关，如果关闭，返回的 CtEntry 中的 chain 为空，从这里可以看出，如果 chain 为空，则不会触发 Sentinel 流控相关的逻辑，从侧面也反应了该属性的重要性。</p>
<p>代码@4：为该资源加载处理链链，这里是最最重要的方法，将在下文详细介绍。</p>
<p>代码@5：根据资源ID、处理器链、上下文环境构建 CtEntry 对象。</p>
<p>代码@6：调用 chain 的 entry 方法。</p>
<p>代码@7：如果出现 BlockException ，调用 CtEntry 的 exit 方法。</p>
<h2 id="2、Sentienl-ProcessorSlot-处理链"><a href="#2、Sentienl-ProcessorSlot-处理链" class="headerlink" title="2、Sentienl ProcessorSlot 处理链"></a>2、Sentienl ProcessorSlot 处理链</h2><p>我们接下来重点看一下 lookProcessChain 方法的实现细节。<br>CtSph#lookProcessChain</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br></pre></td><td class="code"><pre><span class="line"><span class="function">ProcessorSlot&lt;Object&gt; <span class="title">lookProcessChain</span><span class="params">(ResourceWrapper resourceWrapper)</span> </span>&#123;</span><br><span class="line">    ProcessorSlotChain chain = chainMap.get(resourceWrapper);  <span class="comment">// @1</span></span><br><span class="line">    <span class="keyword">if</span> (chain == <span class="keyword">null</span>) &#123;</span><br><span class="line">        <span class="keyword">synchronized</span> (LOCK) &#123;</span><br><span class="line">	    chain = chainMap.get(resourceWrapper);</span><br><span class="line">            <span class="keyword">if</span> (chain == <span class="keyword">null</span>) &#123;</span><br><span class="line">                <span class="comment">// Entry size limit.</span></span><br><span class="line">                <span class="keyword">if</span> (chainMap.size() &gt;= Constants.MAX_SLOT_CHAIN_SIZE) &#123;        <span class="comment">// @2</span></span><br><span class="line">		    <span class="keyword">return</span> <span class="keyword">null</span>;</span><br><span class="line">                &#125;</span><br><span class="line">                chain = SlotChainProvider.newSlotChain();                                      <span class="comment">// @3</span></span><br><span class="line">                Map&lt;ResourceWrapper, ProcessorSlotChain&gt; newMap = <span class="keyword">new</span> HashMap&lt;ResourceWrapper, ProcessorSlotChain&gt;(</span><br><span class="line">                        chainMap.size() + <span class="number">1</span>);</span><br><span class="line">                newMap.putAll(chainMap);</span><br><span class="line">                newMap.put(resourceWrapper, chain);</span><br><span class="line">                chainMap = newMap;</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line">    <span class="keyword">return</span> chain;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>代码@1：chainMap 一个全局的缓存表，即同一个资源 ResourceWrapper (同一个资源名称) 会共同使用同一个 ProcessorSlotChain ，即不同的线程在访问同一个资源保护的代码时，这些线程将共同使用 ProcessorSlotChain  中的各个 ProcessorSlot 。注意留意 ResourceWrapper 的 equals 方法与 hashCode 方法。</p>
<p>代码@2：这里重点想突出，如果同时在进入的资源个数超过 MAX_SLOT_CHAIN_SIZE，默认为 6000，会返回 null，则不对本次请求执行限流，熔断计算，而是直接跳过，这个点还是值得我们注意的。</p>
<p>代码@3：通过 SlotChainProvider 创建对应的处理链。</p>
<p>SlotChainProvider#newSlotChain</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">static</span> ProcessorSlotChain <span class="title">newSlotChain</span><span class="params">()</span> </span>&#123;</span><br><span class="line">	<span class="keyword">if</span> (slotChainBuilder != <span class="keyword">null</span>) &#123;     <span class="comment">// @1</span></span><br><span class="line">		<span class="keyword">return</span> slotChainBuilder.build();</span><br><span class="line">        &#125;</span><br><span class="line">	slotChainBuilder = SpiLoader.loadFirstInstanceOrDefault(SlotChainBuilder.class, DefaultSlotChainBuilder.class);   <span class="comment">// @2</span></span><br><span class="line">	<span class="keyword">if</span> (slotChainBuilder == <span class="keyword">null</span>) &#123;                                                                                                                                        <span class="comment">// @3</span></span><br><span class="line">		RecordLog.warn(<span class="string">&quot;[SlotChainProvider] Wrong state when resolving slot chain builder, using default&quot;</span>);</span><br><span class="line">                slotChainBuilder = <span class="keyword">new</span> DefaultSlotChainBuilder();</span><br><span class="line">         &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">		RecordLog.info(<span class="string">&quot;[SlotChainProvider] Global slot chain builder resolved: &quot;</span></span><br><span class="line">                + slotChainBuilder.getClass().getCanonicalName());</span><br><span class="line">         &#125;</span><br><span class="line">         <span class="keyword">return</span> slotChainBuilder.build();                                                                                                                                   <span class="comment">// @4</span></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>代码@1：如果 slotChainBuilder 不为空，则直接调用其 build 方法构建处理器链。</p>
<p>代码@2：如果为空，首先通过 JAVA 的 SPI 机制，尝试加载自定义的 Slot Chain 构建器实现类。如果需要实现自定义的 Chain 构建器，只需实现  SlotChainBuilder 接口，然后将其放在 classpath 下即可，如果存在多个，以找到的第一个为准。</p>
<p>代码@3：如果从 SPI 机制中加载失败，则使用默认的构建器：DefaultSlotChainBuilder。</p>
<p>代码@4：调用其 build 方法构造 Slot Chain。</p>
<p>那接下来我们先来看看 Sentinel 的 SlotChainBuilder 类体系，然后看看 DefaultSlotChainBuilder 的 build 方法。</p>
<h4 id="2-1-SlotChainBuilder-类体系"><a href="#2-1-SlotChainBuilder-类体系" class="headerlink" title="2.1 SlotChainBuilder  类体系"></a>2.1 SlotChainBuilder  类体系</h4><p><img src="https://img-blog.csdnimg.cn/2020010513491784.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70" alt="在这里插入图片描述"><br>主要有三个实现类，对应热点、接口网关以及普通场景。我们接下来将重点介绍 DefaultSlotChainBuilder ，关于热点限流与网关限流将在后面的文章中详细探讨。</p>
<h4 id="2-2-DefaultSlotChainBuilder-build-方法"><a href="#2-2-DefaultSlotChainBuilder-build-方法" class="headerlink" title="2.2 DefaultSlotChainBuilder build 方法"></a>2.2 DefaultSlotChainBuilder build 方法</h4><p>DefaultSlotChainBuilder#build</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">DefaultSlotChainBuilder</span> <span class="keyword">implements</span> <span class="title">SlotChainBuilder</span> </span>&#123;</span><br><span class="line">    <span class="function"><span class="keyword">public</span> ProcessorSlotChain <span class="title">build</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        ProcessorSlotChain chain = <span class="keyword">new</span> DefaultProcessorSlotChain();</span><br><span class="line">        chain.addLast(<span class="keyword">new</span> NodeSelectorSlot());</span><br><span class="line">        chain.addLast(<span class="keyword">new</span> ClusterBuilderSlot());</span><br><span class="line">        chain.addLast(<span class="keyword">new</span> LogSlot());</span><br><span class="line">        chain.addLast(<span class="keyword">new</span> StatisticSlot());</span><br><span class="line">        chain.addLast(<span class="keyword">new</span> AuthoritySlot());</span><br><span class="line">        chain.addLast(<span class="keyword">new</span> SystemSlot());</span><br><span class="line">        chain.addLast(<span class="keyword">new</span> FlowSlot());</span><br><span class="line">        chain.addLast(<span class="keyword">new</span> DegradeSlot());</span><br><span class="line">        <span class="keyword">return</span> chain;</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>就问大家激不激动，开不开心，从这些 Slot 的名字基本就能得出其含义。</p>
<ul>
<li>NodeSelectorSlot<br>主要用于构建调用链。</li>
<li>ClusterBuilderSlot<br>用于集群限流、熔断。</li>
<li>LogSlot<br>用于记录日志。</li>
<li>StatisticSlot<br>用于实时收集实时消息。</li>
<li>AuthoritySlot<br>用于权限校验的。</li>
<li>SystemSlot<br>用于验证系统级别的规则。</li>
<li>FlowSlot<br>实现限流机制。</li>
<li>DegradeSlot<br>实现熔断机制。</li>
</ul>
          <!--noindex-->
          <div class="post-button text-center">
            <a class="btn" href="/posts/550fbf45.html#more" rel="contents">
              阅读全文 &raquo;
            </a>
          </div>
          <!--/noindex-->
        
      
    </div>
    
    
    

    

    

    

    <footer class="post-footer">
      

      

      

      
      
        <div class="post-eof"></div>
      
    </footer>
  </div>
  
  
  
  </article>


    
      

  

  
  
  

  <article class="post post-type-normal" itemscope itemtype="http://schema.org/Article">
  
  
  
  <div class="post-block">
    <link itemprop="mainEntityOfPage" href="https://www.codingw.net/posts/9ef855bf.html">

    <span hidden itemprop="author" itemscope itemtype="http://schema.org/Person">
      <meta itemprop="name" content="">
      <meta itemprop="description" content="">
      <meta itemprop="image" content="/images/avatar.gif">
    </span>

    <span hidden itemprop="publisher" itemscope itemtype="http://schema.org/Organization">
      <meta itemprop="name" content="中间件兴趣圈">
    </span>

    
      <header class="post-header">

        
        
          <h1 class="post-title" itemprop="name headline">
                
                <a class="post-title-link" href="/posts/9ef855bf.html" itemprop="url">基于 raft 协议的 RocketMQ DLedger 多副本日志复制设计原理</a></h1>
        

        <div class="post-meta">
          <span class="post-time">
            
              <span class="post-meta-item-icon">
                <i class="fa fa-calendar-o"></i>
              </span>
              
                <span class="post-meta-item-text">发表于</span>
              
              <time title="创建于" itemprop="dateCreated datePublished" datetime="2020-09-26T20:14:35+08:00">
                2020-09-26
              </time>
            

            

            
          </span>

          
            <span class="post-category" >
            
              <span class="post-meta-divider">|</span>
            
              <span class="post-meta-item-icon">
                <i class="fa fa-folder-o"></i>
              </span>
              
                <span class="post-meta-item-text">分类于</span>
              
              
                <span itemprop="about" itemscope itemtype="http://schema.org/Thing">
                  <a href="/categories/rocketmq/" itemprop="url" rel="index">
                    <span itemprop="name">rocketmq</span>
                  </a>
                </span>

                
                
              
            </span>
          

          
            
          

          
          
             <span id="/posts/9ef855bf.html" class="leancloud_visitors" data-flag-title="基于 raft 协议的 RocketMQ DLedger 多副本日志复制设计原理">
               <span class="post-meta-divider">|</span>
               <span class="post-meta-item-icon">
                 <i class="fa fa-eye"></i>
               </span>
               
                 <span class="post-meta-item-text">阅读次数&#58;</span>
               
                 <span class="leancloud-visitors-count"></span>
             </span>
          

          

          

          

        </div>
      </header>
    

    
    
    
    <div class="post-body" itemprop="articleBody">

      
      

      
        
          <h2 id="1、RocketMQ-DLedger-多副本日志复制流程图"><a href="#1、RocketMQ-DLedger-多副本日志复制流程图" class="headerlink" title="1、RocketMQ DLedger 多副本日志复制流程图"></a>1、RocketMQ DLedger 多副本日志复制流程图</h2><h3 id="1-1-RocketMQ-DLedger-日志转发-append-请求流程图"><a href="#1-1-RocketMQ-DLedger-日志转发-append-请求流程图" class="headerlink" title="1.1 RocketMQ DLedger 日志转发(append) 请求流程图"></a>1.1 RocketMQ DLedger 日志转发(append) 请求流程图</h3><p><img src="https://img-blog.csdnimg.cn/20190928183406295.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70" alt="在这里插入图片描述"></p>
<h3 id="1-2-RocketMQ-DLedger-日志仲裁流程图"><a href="#1-2-RocketMQ-DLedger-日志仲裁流程图" class="headerlink" title="1.2 RocketMQ DLedger 日志仲裁流程图"></a>1.2 RocketMQ DLedger 日志仲裁流程图</h3><p><img src="https://img-blog.csdnimg.cn/20190928183709529.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70" alt="在这里插入图片描述"></p>
<h3 id="1-3-RocketMQ-DLedger-从节点日志复制流程图"><a href="#1-3-RocketMQ-DLedger-从节点日志复制流程图" class="headerlink" title="1.3 RocketMQ DLedger 从节点日志复制流程图"></a>1.3 RocketMQ DLedger 从节点日志复制流程图</h3><p><img src="https://img-blog.csdnimg.cn/20190928183733569.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70" alt="在这里插入图片描述"></p>
          <!--noindex-->
          <div class="post-button text-center">
            <a class="btn" href="/posts/9ef855bf.html#more" rel="contents">
              阅读全文 &raquo;
            </a>
          </div>
          <!--/noindex-->
        
      
    </div>
    
    
    

    

    

    

    <footer class="post-footer">
      

      

      

      
      
        <div class="post-eof"></div>
      
    </footer>
  </div>
  
  
  
  </article>


    
      

  

  
  
  

  <article class="post post-type-normal" itemscope itemtype="http://schema.org/Article">
  
  
  
  <div class="post-block">
    <link itemprop="mainEntityOfPage" href="https://www.codingw.net/posts/b9d8cad7.html">

    <span hidden itemprop="author" itemscope itemtype="http://schema.org/Person">
      <meta itemprop="name" content="">
      <meta itemprop="description" content="">
      <meta itemprop="image" content="/images/avatar.gif">
    </span>

    <span hidden itemprop="publisher" itemscope itemtype="http://schema.org/Organization">
      <meta itemprop="name" content="中间件兴趣圈">
    </span>

    
      <header class="post-header">

        
        
          <h1 class="post-title" itemprop="name headline">
                
                <a class="post-title-link" href="/posts/b9d8cad7.html" itemprop="url">结合 Sentinel 专栏谈谈我的源码阅读技巧</a></h1>
        

        <div class="post-meta">
          <span class="post-time">
            
              <span class="post-meta-item-icon">
                <i class="fa fa-calendar-o"></i>
              </span>
              
                <span class="post-meta-item-text">发表于</span>
              
              <time title="创建于" itemprop="dateCreated datePublished" datetime="2020-09-25T23:05:35+08:00">
                2020-09-25
              </time>
            

            

            
          </span>

          
            <span class="post-category" >
            
              <span class="post-meta-divider">|</span>
            
              <span class="post-meta-item-icon">
                <i class="fa fa-folder-o"></i>
              </span>
              
                <span class="post-meta-item-text">分类于</span>
              
              
                <span itemprop="about" itemscope itemtype="http://schema.org/Thing">
                  <a href="/categories/sentinel/" itemprop="url" rel="index">
                    <span itemprop="name">sentinel</span>
                  </a>
                </span>

                
                
              
            </span>
          

          
            
          

          
          
             <span id="/posts/b9d8cad7.html" class="leancloud_visitors" data-flag-title="结合 Sentinel 专栏谈谈我的源码阅读技巧">
               <span class="post-meta-divider">|</span>
               <span class="post-meta-item-icon">
                 <i class="fa fa-eye"></i>
               </span>
               
                 <span class="post-meta-item-text">阅读次数&#58;</span>
               
                 <span class="leancloud-visitors-count"></span>
             </span>
          

          

          

          

        </div>
      </header>
    

    
    
    
    <div class="post-body" itemprop="articleBody">

      
      

      
        
          <blockquote>
<p>本文行文思路：先抛出源码阅读方法，然后结合Sentinel创作过程谈谈具体是如何运用这些技巧，最后解答几个源码阅读的误区。</p>
</blockquote>
<p>Sentinel 系列共包含15篇文章，主要以源码分析为手段，图文并茂的方式对 Sentinel 的架构设计理念、核心实现要点进行了一一剖析，并加以实战分析与思考。</p>
<p><strong>很多朋友都在询我是如何阅读源码的。对此可归纳为如下几个要点，然后结合 Sentienl 源码分析专栏对各个要点进行拆解，对源码阅读方法进行一次“实战”。</strong></p>
<ul>
<li>阅读官方文档，从全局了解待学习框架能解决什么样的问题，整体架构设计与思想是什么，主要包含哪些要点。</li>
<li>从官方提供的 Demo 程序开始，学习基本的使用方法，进一步加深其理解，并伺机寻找入口（突破口）。</li>
<li>寻找突破口，逐一突破，先主干再旁支，适度分解，各个击破。</li>
</ul>
<p>接下来将展示我是如何使用这套方法论来学习 Sentinel 的。</p>
<p>在准备深入学习 Sentinel 之前，首先认真看了一遍 Sentinel 的官方文档，从而形成了对 Sentinel 的基本认识，我们可以从官方文档了解到 Sentinel 主要涉及的核心内容，正如下图所示：<br><img src="https://img-blog.csdnimg.cn/2020053015083815.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述">这些内容可以简单的当成一个学习的提纲，可以从里面挑选自己感兴趣的进行重点学习。</p>
<p>在看完官方文档后，我就踏上创作 Sentinel 系列的征途。</p>
<blockquote>
<p>本文不会再详细介绍每一个知识点的细节，有关各个知识点的具体讲解，大家可以点击感兴趣的链接中进行深入研究，本文主要是点到为止，重在介绍笔者是如何学习源码的。</p>
</blockquote>
<p>1、<a target="_blank" rel="noopener" href="https://blog.csdn.net/prestigeding/article/details/103544443">Alibaba Sentinel 限流与熔断初探</a><br>该文章主要从如下几个点进行展开：</p>
<ul>
<li>Sentinel 是什么 ？主要能解决什么问题？</li>
<li>限流与熔断的使用场景</li>
<li>Sentinel 源码结构</li>
<li>在 IntelliJ IDEA 中运行 Sentine Demo</li>
</ul>
<p>其实第三点并不是特别必须，不过要得出这些结论也并不难，因为对 Sentinel 有了全局的认识后并根据各个模块的命名很容易能得出该模块的作用。这里第四点非常关键，通常一个优秀的开源框架都会提供完备的演示 Demo，大家可以看到 Sentinel 的演示 demo 非常丰富，在本文中我特意选择了 Dubbo 来做示例，主要是我们公司大量使用 dubbo 来实现公司的微服务，这样会更加贴近实战，更有利于寻找突破口。</p>
<p>通过跑通 Demo 的主要目的有三个：</p>
<ul>
<li>通过运行 Demo，了解框架的基本使用方法。</li>
<li>搭建一个可 Debug 的环境，为后续看不懂代码的情况下进行调试，根据运行时数据，可加快代码的理解速度，但千万不要一开始就 debug。</li>
<li>寻找源码阅读的入口。</li>
</ul>
<p>2、<a target="_blank" rel="noopener" href="https://blog.csdn.net/prestigeding/article/details/103654590">源码分析 Sentinel 之 Dubbo 适配原理</a><br>紧跟第一篇文章，既然使用的是 Dubbo 作为其示例代码，自然而然的思考 Sentinel 是如何做到对 Dubbo 的适配并对业务无侵入性。</p>
<p>通过该篇文章的学习我们了解到可以通过 Dubbo 的扩展机制实现对 Dubbo 的适配，在 Dubbo Filter 中我们能看到了与 Sentinel 相关的核心 API SphU.entry，从而找到深入学习 Sentinel 的核心入口，也就是后续文章会通过对该方法的研究，从而打开进入 Sentinel 内核世界的大门。</p>
<blockquote>
<p>备注：在阅读这篇文章的时候，我觉得 Dubbo 的适配感觉非常简单，但随着我对这个系列的深入学习，发现了该方法没有那么简单，当时很多点都没有理解到位，这个在后续会有重点阐述，这也是不断学习、不断思考带来的好处。</p>
</blockquote>
          <!--noindex-->
          <div class="post-button text-center">
            <a class="btn" href="/posts/b9d8cad7.html#more" rel="contents">
              阅读全文 &raquo;
            </a>
          </div>
          <!--/noindex-->
        
      
    </div>
    
    
    

    

    

    

    <footer class="post-footer">
      

      

      

      
      
        <div class="post-eof"></div>
      
    </footer>
  </div>
  
  
  
  </article>


    
      

  

  
  
  

  <article class="post post-type-normal" itemscope itemtype="http://schema.org/Article">
  
  
  
  <div class="post-block">
    <link itemprop="mainEntityOfPage" href="https://www.codingw.net/posts/97996927.html">

    <span hidden itemprop="author" itemscope itemtype="http://schema.org/Person">
      <meta itemprop="name" content="">
      <meta itemprop="description" content="">
      <meta itemprop="image" content="/images/avatar.gif">
    </span>

    <span hidden itemprop="publisher" itemscope itemtype="http://schema.org/Organization">
      <meta itemprop="name" content="中间件兴趣圈">
    </span>

    
      <header class="post-header">

        
        
          <h1 class="post-title" itemprop="name headline">
                
                <a class="post-title-link" href="/posts/97996927.html" itemprop="url">我的另一种参与 RocketMQ 开源社区的方式</a></h1>
        

        <div class="post-meta">
          <span class="post-time">
            
              <span class="post-meta-item-icon">
                <i class="fa fa-calendar-o"></i>
              </span>
              
                <span class="post-meta-item-text">发表于</span>
              
              <time title="创建于" itemprop="dateCreated datePublished" datetime="2020-09-24T22:49:57+08:00">
                2020-09-24
              </time>
            

            

            
          </span>

          
            <span class="post-category" >
            
              <span class="post-meta-divider">|</span>
            
              <span class="post-meta-item-icon">
                <i class="fa fa-folder-o"></i>
              </span>
              
                <span class="post-meta-item-text">分类于</span>
              
              
                <span itemprop="about" itemscope itemtype="http://schema.org/Thing">
                  <a href="/categories/rocketmq/" itemprop="url" rel="index">
                    <span itemprop="name">rocketmq</span>
                  </a>
                </span>

                
                
              
            </span>
          

          
            
          

          
          
             <span id="/posts/97996927.html" class="leancloud_visitors" data-flag-title="我的另一种参与 RocketMQ 开源社区的方式">
               <span class="post-meta-divider">|</span>
               <span class="post-meta-item-icon">
                 <i class="fa fa-eye"></i>
               </span>
               
                 <span class="post-meta-item-text">阅读次数&#58;</span>
               
                 <span class="leancloud-visitors-count"></span>
             </span>
          

          

          

          

        </div>
      </header>
    

    
    
    
    <div class="post-body" itemprop="articleBody">

      
      

      
        
          <p>首先先“SHOW”一波我在 RocketMQ 开源社区所获得的成就：2019年RocketMQ社区授予我优秀布道师荣誉称号，证书很高大上，奖品丰厚哦。<br><img src="https://img-blog.csdnimg.cn/20200608190532198.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述"><br>布道师是什么？开源项目不是都在追求如何成为一名 Committer？其实这个就是参与开源项目的两种不同方式。如何参与一个开源项目，容我慢慢道来。</p>
<p>@<a href="%E6%9C%AC%E8%8A%82%E7%9B%AE%E5%BD%95">TOC</a></p>
<h2 id="1、与-RocketMQ-相识、相知到“在一起”"><a href="#1、与-RocketMQ-相识、相知到“在一起”" class="headerlink" title="1、与 RocketMQ 相识、相知到“在一起”"></a>1、与 RocketMQ 相识、相知到“在一起”</h2><p>在2017年听到阿里巴巴将 RocketMQ 捐赠给 Apache基金会成为 Apache 的顶级项目，我内心是无比激动，因为终于可以一睹一款高性能的消息中间件的实现原理。</p>
<p>通过阅读 RocketMQ 官方文档，以下几个特别的点吸引了我的注意，让我下定决心深入研究一番。</p>
<ul>
<li>RocketMQ 为什么性能高效，到底运用了什么“厉害”的技术</li>
<li>RocketMQ 如何实现刷盘（可以类比一下数据库方面的刷盘、redo、undo日志）</li>
<li>RocketMQ 文件存储设计理念、基于文件的 Hash 索引是怎么实现的</li>
<li>定时消息、消息过滤等实现原理</li>
<li>如何进行网络编程（Netty实战）</li>
</ul>
<p>心动不如行动，下定决心后便开始了我的源码分析 RocketMQ 之旅，大概在4个多月的时间中连续发表了30余篇文章，从 Nameserver、消息发送高可用设计、消息存储、消息消费、消息过滤、事务消息等各个方面对其进行了体系化的剖析，边写边分享，边分享边传播，终于得到了机械工业出版社华章分社的杨福川老师的认可，邀请我出书。</p>
<p>在杨老师和张工的帮助与指点下，经过将近半年的努力，书稿基本完稿。</p>
<p>由于我当时是一位名不经传的新人，按照出版行业的惯例，需要找一些该领域内专家大牛帮忙做序或写写推荐语。</p>
<p>当时我是初生牛犊不怕虎，蹦出了一个非常大胆的想法，是不是可以联系 RocketMQ 官方的一些大佬，最终我直接锁定了 RocketMQ 创始人冯嘉大神，希望他能帮我作序推荐。</p>
<p>令人惊喜的是冯嘉大神非常平易见人，得知我的来意后，他说了这样一句话：“我是非常愿意为写书的朋友作序，但需要评估一下书稿的质量，如果质量OK，非常愿意效劳”。</p>
<p>我备受鼓舞，在和出版社初步沟通后，将试读稿件再加上消息存储整章的内容发给冯嘉大神后，经冯嘉大神认真审稿后，决定帮忙推荐作序，真的让我备受鼓舞。</p>
<p>随着《RocketMQ技术内幕》一书的正式出版上市，并得到广大读者朋友的认可，与官方的联系也越来越多，后面在 RocketMQ 中国社区负责人青峰大佬的筹备下，我还参与了 RocketMQ 官方社区的源码解析直播活动、官方文档审稿等工作，并在社区得到了不错的反响。</p>
<p><strong>说到这里大家是不是觉得非常奇怪，是不是都认为你只是在写文章，写书，没有真正参与开源社区呀，没有贡献代码，这个算哪门子参与开源社区？</strong></p>
<p>其实我一开始连我自己也没有意识到我正在参与一个开源项目，直到我在冯嘉大神为我写的序言中看到他给了我一个新的称号：<strong>RocketMQ布道师</strong>，从而才真正了解到参与开源的另外一种方式：做一个开源项目的传播者，让更多人更容易的使用它，即降低大众对它的使用门槛。</p>
<p>我后面也特意去查了一下开源项目的布道师是一个什么的角色，或者说什么样的人能被称之为布道师。个人的理解就是首先认可并热爱这项技术，并持续输出高质量的技术类文章、文档等有助于技术传播的素材，让更多人更容易理解并使用它。</p>
<p>有了新的称号，那就得更加努力，朝着优秀努力，在2019年我又陆续发表了20几篇关于RocketMQ相关的文章，这些文章含金量极高，不仅及时跟进了RocketMQ4.3之后的新特性：消息轨迹、ACL、主从切换等机制，更是发表了数篇实战类文章，详细指出在生产环境下一些使用误区，更是输出了几篇生产环境真实故障与解决方案。最终于2019年 RocketMQ 官方社区授予我优秀布道师荣誉称号。</p>
<p>RocketMQ 成就了现在的我，我也会继续努力，为传播RocketMQ尽一份力所能及的力量。2020年，继续努力。</p>
<h2 id="2、如何成为开源项目的-Committer"><a href="#2、如何成为开源项目的-Committer" class="headerlink" title="2、如何成为开源项目的 Committer"></a>2、如何成为开源项目的 Committer</h2><p>有一些粉丝在问我，您对 RocketMQ 研究的这么深入，为什么不考虑贡献代码，成为一名 Committer 呢？这是因为参与开源项目需要具备一些基本条件，当下我的实际情况不符合，那成为一个开源项目的 Committer 有些什么条件呢？</p>
          <!--noindex-->
          <div class="post-button text-center">
            <a class="btn" href="/posts/97996927.html#more" rel="contents">
              阅读全文 &raquo;
            </a>
          </div>
          <!--/noindex-->
        
      
    </div>
    
    
    

    

    

    

    <footer class="post-footer">
      

      

      

      
      
        <div class="post-eof"></div>
      
    </footer>
  </div>
  
  
  
  </article>


    
      

  

  
  
  

  <article class="post post-type-normal" itemscope itemtype="http://schema.org/Article">
  
  
  
  <div class="post-block">
    <link itemprop="mainEntityOfPage" href="https://www.codingw.net/posts/a815ae71.html">

    <span hidden itemprop="author" itemscope itemtype="http://schema.org/Person">
      <meta itemprop="name" content="">
      <meta itemprop="description" content="">
      <meta itemprop="image" content="/images/avatar.gif">
    </span>

    <span hidden itemprop="publisher" itemscope itemtype="http://schema.org/Organization">
      <meta itemprop="name" content="中间件兴趣圈">
    </span>

    
      <header class="post-header">

        
        
          <h1 class="post-title" itemprop="name headline">
                
                <a class="post-title-link" href="/posts/a815ae71.html" itemprop="url">Sentinel 集群限流设计原理</a></h1>
        

        <div class="post-meta">
          <span class="post-time">
            
              <span class="post-meta-item-icon">
                <i class="fa fa-calendar-o"></i>
              </span>
              
                <span class="post-meta-item-text">发表于</span>
              
              <time title="创建于" itemprop="dateCreated datePublished" datetime="2020-09-23T23:03:35+08:00">
                2020-09-23
              </time>
            

            

            
          </span>

          
            <span class="post-category" >
            
              <span class="post-meta-divider">|</span>
            
              <span class="post-meta-item-icon">
                <i class="fa fa-folder-o"></i>
              </span>
              
                <span class="post-meta-item-text">分类于</span>
              
              
                <span itemprop="about" itemscope itemtype="http://schema.org/Thing">
                  <a href="/categories/sentinel/" itemprop="url" rel="index">
                    <span itemprop="name">sentinel</span>
                  </a>
                </span>

                
                
              
            </span>
          

          
            
          

          
          
             <span id="/posts/a815ae71.html" class="leancloud_visitors" data-flag-title="Sentinel 集群限流设计原理">
               <span class="post-meta-divider">|</span>
               <span class="post-meta-item-icon">
                 <i class="fa fa-eye"></i>
               </span>
               
                 <span class="post-meta-item-text">阅读次数&#58;</span>
               
                 <span class="leancloud-visitors-count"></span>
             </span>
          

          

          

          

        </div>
      </header>
    

    
    
    
    <div class="post-body" itemprop="articleBody">

      
      

      
        
          
            <div id="vip-container"><h2 id="1、集群限流使用场景"><a href="#1、集群限流使用场景" class="headerlink" title="1、集群限流使用场景"></a>1、集群限流使用场景</h2><p>首先一个服务有三个服务提供者，但这三台集群的硬件配置不一样，如图所示：<br><img src="https://img-blog.csdnimg.cn/20200503232733111.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述"><br>为了充分利用硬件的资源，诸如 Dubbo 都提供了基于权重的负载均衡机制，例如可以将8C16G的机器设置的权重是4C8G的两倍，这样充分利用硬件资源，假如现在需要引入 Sentinel 的限流机制，例如为一个 Dubbo 服务设置限流规则，这样由于三台集群分担的流量不均匀，会导致无法重复利用高配机器的资源。</p>
<p>假设经过压测，机器配置为C48G最高能承受的TPS为 1500，而机器配置为8C16G能承受的TPS为2800，那如果采取单机限流，其阔值只能设置为1500，因为如果超过1500，会将4C8G的机器压垮。</p>
<p>解决这种办法的方式就是针对整个集群进行限流，即为整个集群设置一个阔值，例如设置限流TPS为6000。</p>
<h2 id="2、集群限流与单机限流的异同思考"><a href="#2、集群限流与单机限流的异同思考" class="headerlink" title="2、集群限流与单机限流的异同思考"></a>2、集群限流与单机限流的异同思考</h2><p>限流的一个基本作用就是按照限流规则生成访问许可(Token)，然后根据当前实时的调用信息进行判断是否可以获得许可而决定是否放行。</p>
<p>集群与单机限流在实时调用信息收集方面应该差别不大，都可以基于滑动窗口进行统计信息的收集。</p>
<p>集群与单机限流的最主要区别在与许可的生成，单机模式的许可直接在本地生成，但集群限流必须有一个统一的 Token 发放机制，以此来协调当前集群内多机调用，从而基于当前“调用总数”进行限流。</p>
<h2 id="3、探究集群限流实现原理"><a href="#3、探究集群限流实现原理" class="headerlink" title="3、探究集群限流实现原理"></a>3、探究集群限流实现原理</h2><p>在探究集群限流实现原理之前先来回顾一下单机限流的执行流程图。<br><img src="https://img-blog.csdnimg.cn/20200503232832535.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述"><br>结合流程我们可以看出集群限流的几个关键点 ClusterBuilderSlot、FlowSlot。</p>
<h4 id="3-1-ClusterBuilderSlot-详解"><a href="#3-1-ClusterBuilderSlot-详解" class="headerlink" title="3.1 ClusterBuilderSlot 详解"></a>3.1 ClusterBuilderSlot 详解</h4><p>在对一个资源进行流控规则判断时，首先将进入到 NodeSelectorSlot，然后就会进入到 ClusterBuilderSlot，为了与单机限流模式，介绍 ClusterBuilderSlot 时与 NodeSelectorSlot 进行一个对比。</p>
<p>NodeSelectorSlot 的核心实现截图如下所示：<br><img src="https://img-blog.csdnimg.cn/20200503232919521.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述"></p>
<blockquote>
<p>温馨提示：从该系列之前的文章也能得知，一个 资源对应的一个 NodeSelectorSlot  实例，即多线程访问一个资源时，都会调用同一个 NodeSelectorSlot 实例。</p>
</blockquote>
<p>NodeSelectorSlot 的关键点如下：</p>
<ul>
<li>Map&lt;String, DefaultNode&gt; map<br>在 NodeSelectorSlot 中是以 context Id 为维度进行缓存的，例如官方给出的 Dubbo 适配方法，contexId 为 dubbo 服务的全路径名。即 Dubbo的入口节点对应的缓存 Key 为 context<br>id。</li>
<li>fireEntry 的 node 参数<br>由于 NodeSelectorSlot 是第一个过滤器，故第一次调用 fireEntry 方法时的 node 参数就是上面创建的 Node，即与 context 相关链的 Node，即所谓的入口节点即 Entrance Node。</li>
</ul>
<p>接下来重点关注一下 ClusterBuilderSlot 的关键点：<br><img src="https://img-blog.csdnimg.cn/20200503233012141.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述"><br>ClusterBuilderSlot 的关键点如下：</p>
<ul>
<li>Map&lt;ResourceWrapper, ClusterNode&gt; clusterNodeMap<br>持有的集群节点缓存表，其键为 Entrance Node 所对应的资源ID，即 Context 中关联的节点信息。</li>
<li>Node originNode<br>所谓的 orginNode，即在调用 ContextUtil 中 enter(String name, String origin) 方法中的第二个参数，表示这条调用链的源头，在 Dubbo 中默认为 应用的 application。</li>
</ul>
<p>经过上面两个Slot，整个调用链就基本创建好了，接下来我们来看一下 FlowSlot 关于集群限流的相关处理逻辑。</p>
<h4 id="3-2-集群限流模式实现原理"><a href="#3-2-集群限流模式实现原理" class="headerlink" title="3.2 集群限流模式实现原理"></a>3.2 集群限流模式实现原理</h4><p><img src="https://img-blog.csdnimg.cn/20200503233045696.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述"><br>FlowSlow FlowSlot 的核心处理逻辑主要是调用 FlowRuleChecker 的 canPassCheck 方法，正如上面看到的一样，根据配置规则，如果是集群模式，则调用的是其 passClusterCheck 方法，接下来我们将重点探讨该方法。<br><img src="https://img-blog.csdnimg.cn/20200503233111258.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="FlowRuleChecker#passClusterCheck"><br>代码@1：获取一个 TokenService 服务类。这里实现关键点：</p>
<ul>
<li>如果当前节点的角色为 CLIENT，返回的 TokenService 为 DefaultClusterTokenClient。</li>
<li>如果当前节点的角色为 SERVER，返回的 TokenService 为 ClusterTokenServer，这里使用了SPI极致，可以通过查看 META-INF/services 目录下的 com.alibaba.csp.sentinel.cluster.TokenService 文件，默认服务端返回 DefaultTokenService。</li>
</ul>
<p>代码@2：如果无法获取到集群限流Token服务，如果该限流规则配置了可以退化为单机限流模式，则退化为单机限流。</p>
<p>代码@3：获取集群限流的流程ID，该 flowId 全局唯一。</p>
<p>代码@4：通过 TokenService 去申请 token，这里是与单机限流模式最大的差别。</p>
<p>接下来将分别从 DefaultClusterTokenClient、DefaultTokenService 分别探究集群限流相关的实现原理与细节，更好的指导我们如何使用集群限流功能。</p>
<h5 id="3-2-1-DefaultClusterTokenClient-详解"><a href="#3-2-1-DefaultClusterTokenClient-详解" class="headerlink" title="3.2.1 DefaultClusterTokenClient 详解"></a>3.2.1 DefaultClusterTokenClient 详解</h5><p>从我们的经验也得知，TokenClient 的主要职责就是发送请求到 TokenService 端，主要是网络相关的细节将不在此篇文章中给出，如果有兴趣，大家可以关注我的 Netty 专栏。</p>
<p>首先 Sentinel 提供了 SPI 机制，故允许用户自定义 TokenClient 的实现类，官方与 SPI 默认配置的文件如下：<br><img src="https://img-blog.csdnimg.cn/2020050323322512.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述"><br>关于 TokenClient 主要关注其初始化代码，因为我们需要关注一个非常重要的点：<br>DefaultClusterTokenClient#initNewConnection<br><img src="https://img-blog.csdnimg.cn/20200503233259244.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述"><br>在客户端启动的时候会创建与 TokenServer 之间的链接，即这边需要配置服务端的 IP 与端口号，那如何配置呢？其实配置方式完全由自己去实现对应的解析器，下面根据官方的 Demo 示例如下：<br><img src="https://img-blog.csdnimg.cn/20200503233319545.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述"><br>这里需要说明的其配置项由 ClusterGroupEntity 来定义，其字段的定义如下：</p>
<ul>
<li>clientSet<br>客户端 Set 集合。</li>
<li>ip<br>Token 服务端的 IP。</li>
<li>machinedId<br>Token 服务端的机器ID。</li>
<li>port<br>Token 服务端的机器端口。</li>
</ul>
<p>其配置示例如下：</p>
<figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">[&#123;&quot;clientSet&quot;:[&quot;112.12.88.66@8729&quot;,&quot;112.12.88.67@8727&quot;],&quot;ip&quot;:&quot;112.12.88.68&quot;,&quot;machineId&quot;:&quot;112.12.88.68@8728&quot;,&quot;port&quot;:11111&#125;]</span><br></pre></td></tr></table></figure>

<p>Client 端接下来就是向服务端发送请求，与网络相关的不在本文的讨论范围内，接下来将重点探讨服务端是如何发放许可的。</p>
<h5 id="3-2-2-DefaultTokenService-详解"><a href="#3-2-2-DefaultTokenService-详解" class="headerlink" title="3.2.2 DefaultTokenService 详解"></a>3.2.2 DefaultTokenService 详解</h5><p>Token Server 端收到客户的请求，其处理入口为 FlowRequestProcessor，其处理方法为：processRequest，最终会调用 DefaultTokenService 的 requestToken 方法。<br>DefaultTokenService#requestToken<br><img src="https://img-blog.csdnimg.cn/20200503233418165.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述"><br>代码@1：根据 ruleId 获取指定的限流规则。</p>
<p>代码@2：然后调用 ClusterFlowChecker 的 acquierClusterToken 方法，申请许可。</p>
<p>许可的发放流程主要由 ClusterFlowChecker 的 acquierClusterToken 方法实现。<br><img src="https://img-blog.csdnimg.cn/20200503233520315.png#pic_center" alt="在这里插入图片描述"><br>Step1：首先判断是否允许本次许可申请，这是因为 TokenServe 支持嵌入式，即支持在应用节点中嵌入一个 TokenServer，为了保证许可申请的请求不对正常业务造成比较大的影响，故对申请许可这个动作进行了限流。</p>
<p>一旦触发了限流，将向客户端返回 TOO_MANY_REQUEST 状态码，Sentinel 支持按 namespace 进行限流，具体由 GlobalRequestLimiter 实现，该类的内部同样基于滑动窗口进行收集，原理与 FlowSlot 相似，故这里不加以展开，默认的限流TPS为3W，有关于 Sentinel 相关的配置，将在后续文章专门梳理。<br><img src="https://img-blog.csdnimg.cn/20200503233607726.png#pic_center" alt="在这里插入图片描述"><br>Step2：根据流程ID获取指标采集器。<br><img src="https://img-blog.csdnimg.cn/20200503233630847.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述"><br>Step3：计算 latestQps、globalThreashold、 nextRemaining 三个阔值，三个的含义分别如下：</p>
<ul>
<li>latestQps<br>获取当前正常访问的QPS。</li>
<li>globalThreashold<br>根据限流配置规则得出其总许可数量，其主要根据阔值的方式其有所不同，其配置阔值有两种方式：<br>1）FLOW_THRESHOLD_GLOBAL<br>总数，即集群中的许可等于限流规则中配置的 count 值。<br>2）FLOW_THRESHOLD_AVG_LOCAL<br>单机分摊模式，此时限流规则中配置的值只是单机的 count 值，集群中的许可数等于 count * 集群中客户端的个数。<br>注意：这里还可以通过 exceedCount 设置来运行超过其最大阔值，默认为1表示不允许超过。</li>
<li>nextRemainging<br>表示处理完本次请求后剩余的许可数量。<br><img src="https://img-blog.csdnimg.cn/20200503233707407.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述">Step4：如果剩余的许可数大于0，则本次申请成功，将当前的调用计入指标采集器中，然后返回给客户即可。</li>
</ul>
<p>接下来所有流程步骤都是基于没有剩余许可数的处理逻辑。<br><img src="https://img-blog.csdnimg.cn/20200503233730680.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述"><br>Step5：当前许可数不足的情况，并且该请求为高优先级的处理逻辑：</p>
<ul>
<li>获取当前等待的TPS（即1s为维度，当前等待的请求数量）</li>
<li>如果当前等待的TPS低于可借用未来窗口的许可阔值时，可通过，但设置其等待时间，可以通过 maxOccupyRatio 来设置借用的最大比值。<br><img src="https://img-blog.csdnimg.cn/20200503233758997.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述">Step6：如果当前许可不足，并且该请求为普通优先级的处理逻辑，增加阻塞相关指标的统计数，并返回 BLOCKED。</li>
</ul>
<p>TokenServer 返回申请许可之后，那 Token Client 如何处理呢？其处理代码在 FlowRuleChecker#applyTokenResult<br><img src="https://img-blog.csdnimg.cn/20200503233827875.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述"><br>我们可以发现，如果服务端返回OK，则顺利通过，返回BLOCKED，则直接返回 false，会抛出 FlowException，如果是 token 限流，如果规则运行退化为单机限流模式，则进行单机限流。</p>
<p>集群限流的基本实现原理就介绍到这里了。</p>
<h2 id="4、总结"><a href="#4、总结" class="headerlink" title="4、总结"></a>4、总结</h2><p>集群限流的基本原理接介绍到这里了，与单机限流模式最大的区别就是集群限流模式的需要引入 TokenService，提供许可的发放服务，该服务可以嵌入应用节点，也可以独立于应用之外。这边借用官方文档上的两张图来简单介绍一下嵌入模式与独立模式的架构：<br><img src="https://img-blog.csdnimg.cn/20200503233858200.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70" alt="在这里插入图片描述"><br><img src="https://img-blog.csdnimg.cn/20200503233858165.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70" alt="在这里插入图片描述"><br>集群模式使用注意，如果使用的是集群模式限流，则如下两个配置则失效：<br><img src="https://img-blog.csdnimg.cn/20200503233920874.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ByZXN0aWdlZGluZw==,size_16,color_FFFFFF,t_70#pic_center" alt="在这里插入图片描述"><br><strong>最后抛出一个思考题：集群模式应该算是高大上，但我们项目中真的需要吗？集群限流模式有哪些缺点、哪些优点，欢迎大家留言探讨。</strong></p>
</div>

			<script src="https://my.openwrite.cn/js/readmore.js" type="text/javascript"></script>
			<script>
			var isMobile = navigator.userAgent.match(/(phone|pad|pod|iPhone|iPod|ios|iPad|Android|Mobile|BlackBerry|IEMobile|MQQBrowser|JUC|Fennec|wOSBrowser|BrowserNG|WebOS|Symbian|Windows Phone)/i);
			if (!isMobile) {
			    var btw = new BTWPlugin();
			    btw.init({
			        "id": "vip-container",
			        "blogId": "18019-1573088808868-542",
			        "name": "中间件兴趣圈",
			        "qrcode": "https://img-blog.csdnimg.cn/20190314214003962.jpg",
			        "keyword": "more"
			    });
			}
			</script>
		
          
        
      
    </div>
    
    
    

    

    

    

    <footer class="post-footer">
      

      

      

      
      
        <div class="post-eof"></div>
      
    </footer>
  </div>
  
  
  
  </article>


    
      

  

  
  
  

  <article class="post post-type-normal" itemscope itemtype="http://schema.org/Article">
  
  
  
  <div class="post-block">
    <link itemprop="mainEntityOfPage" href="https://www.codingw.net/posts/5f8f7185.html">

    <span hidden itemprop="author" itemscope itemtype="http://schema.org/Person">
      <meta itemprop="name" content="">
      <meta itemprop="description" content="">
      <meta itemprop="image" content="/images/avatar.gif">
    </span>

    <span hidden itemprop="publisher" itemscope itemtype="http://schema.org/Organization">
      <meta itemprop="name" content="中间件兴趣圈">
    </span>

    
      <header class="post-header">

        
        
          <h1 class="post-title" itemprop="name headline">
                
                <a class="post-title-link" href="/posts/5f8f7185.html" itemprop="url">源码分析 Sentinel 实时数据采集实现原理</a></h1>
        

        <div class="post-meta">
          <span class="post-time">
            
              <span class="post-meta-item-icon">
                <i class="fa fa-calendar-o"></i>
              </span>
              
                <span class="post-meta-item-text">发表于</span>
              
              <time title="创建于" itemprop="dateCreated datePublished" datetime="2020-09-22T22:47:35+08:00">
                2020-09-22
              </time>
            

            

            
          </span>

          
            <span class="post-category" >
            
              <span class="post-meta-divider">|</span>
            
              <span class="post-meta-item-icon">
                <i class="fa fa-folder-o"></i>
              </span>
              
                <span class="post-meta-item-text">分类于</span>
              
              
                <span itemprop="about" itemscope itemtype="http://schema.org/Thing">
                  <a href="/categories/sentinel/" itemprop="url" rel="index">
                    <span itemprop="name">sentinel</span>
                  </a>
                </span>

                
                
              
            </span>
          

          
            
          

          
          
             <span id="/posts/5f8f7185.html" class="leancloud_visitors" data-flag-title="源码分析 Sentinel 实时数据采集实现原理">
               <span class="post-meta-divider">|</span>
               <span class="post-meta-item-icon">
                 <i class="fa fa-eye"></i>
               </span>
               
                 <span class="post-meta-item-text">阅读次数&#58;</span>
               
                 <span class="leancloud-visitors-count"></span>
             </span>
          

          

          

          

        </div>
      </header>
    

    
    
    
    <div class="post-body" itemprop="articleBody">

      
      

      
        
          <p>本篇将重点关注 Sentienl 实时数据收集，即 Sentienl 具体是如何收集调用信息，以此来判断是否需要触发限流或熔断。</p>
<p>Sentienl 实时数据收集的入口类为 StatisticSlot。</p>
<p>我们先简单来看一下 StatisticSlot 该类的注释，来看一下该类的整体定位。</p>
<p>StatisticSlot，专用于实时统计的 slot。在进入一个资源时，在执行 Sentienl 的处理链条中会进入到该 slot 中，需要完成如下计算任务：</p>
<ul>
<li>集群维度计算资源的总统计信息，用于集群限流，后续文章将详细探讨。</li>
<li>来自不同调用方/来源的群集节点的统计信息。</li>
<li>特定调用上下文环境的统计信息。</li>
<li>统计所有入口的统计信息。</li>
</ul>
<p>接下来用源码分析的手段来详细分析 StatisticSlot 的实现原理。</p>
<h2 id="1、源码分析-StatisticSlot"><a href="#1、源码分析-StatisticSlot" class="headerlink" title="1、源码分析 StatisticSlot"></a>1、源码分析 StatisticSlot</h2><h4 id="1-1-StatisticSlot-entry-详解"><a href="#1-1-StatisticSlot-entry-详解" class="headerlink" title="1.1 StatisticSlot entry 详解"></a>1.1 StatisticSlot entry 详解</h4><p>StatisticSlot#entry </p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">entry</span><span class="params">(Context context, ResourceWrapper resourceWrapper, DefaultNode node, <span class="keyword">int</span> count,<span class="keyword">boolean</span> prioritized, Object... args)</span> <span class="keyword">throws</span> Throwable </span>&#123; </span><br><span class="line">	<span class="keyword">try</span> &#123;</span><br><span class="line">		<span class="comment">// Do some checking.</span></span><br><span class="line">                fireEntry(context, resourceWrapper, node, count, prioritized, args);  <span class="comment">// @1</span></span><br><span class="line">        	<span class="comment">// Request passed, add thread count and pass count.</span></span><br><span class="line">        	node.increaseThreadNum();                                                             <span class="comment">// @2</span></span><br><span class="line">       		node.addPassRequest(count);</span><br><span class="line">		<span class="keyword">if</span> (context.getCurEntry().getOriginNode() != <span class="keyword">null</span>) &#123;                           <span class="comment">// @3</span></span><br><span class="line">			<span class="comment">// Add count for origin node.</span></span><br><span class="line">            		context.getCurEntry().getOriginNode().increaseThreadNum();</span><br><span class="line">            		context.getCurEntry().getOriginNode().addPassRequest(count);</span><br><span class="line">       		 &#125;</span><br><span class="line">		<span class="keyword">if</span> (resourceWrapper.getEntryType() == EntryType.IN) &#123;                <span class="comment">// @4</span></span><br><span class="line">			<span class="comment">// Add count for global inbound entry node for global statistics.</span></span><br><span class="line">            		Constants.ENTRY_NODE.increaseThreadNum();</span><br><span class="line">           	 	Constants.ENTRY_NODE.addPassRequest(count);</span><br><span class="line">        	&#125;</span><br><span class="line">		<span class="comment">// Handle pass event with registered entry callback handlers.</span></span><br><span class="line">        	<span class="keyword">for</span> (ProcessorSlotEntryCallback&lt;DefaultNode&gt; handler : StatisticSlotCallbackRegistry.getEntryCallbacks()) &#123;   <span class="comment">// @5</span></span><br><span class="line">            		handler.onPass(context, resourceWrapper, node, count, args);</span><br><span class="line">        	&#125;</span><br><span class="line">    	&#125; <span class="keyword">catch</span> (PriorityWaitException ex) &#123;                                                                                                                                <span class="comment">// @6</span></span><br><span class="line">		node.increaseThreadNum();</span><br><span class="line">		<span class="keyword">if</span> (context.getCurEntry().getOriginNode() != <span class="keyword">null</span>) &#123;</span><br><span class="line">			<span class="comment">// Add count for origin node.</span></span><br><span class="line">            		context.getCurEntry().getOriginNode().increaseThreadNum();</span><br><span class="line">       	 	&#125;</span><br><span class="line">		<span class="keyword">if</span> (resourceWrapper.getEntryType() == EntryType.IN) &#123;</span><br><span class="line">			<span class="comment">// Add count for global inbound entry node for global statistics.</span></span><br><span class="line">            		Constants.ENTRY_NODE.increaseThreadNum();</span><br><span class="line">        	&#125;</span><br><span class="line">        	<span class="comment">// Handle pass event with registered entry callback handlers.</span></span><br><span class="line">        	<span class="keyword">for</span> (ProcessorSlotEntryCallback&lt;DefaultNode&gt; handler : StatisticSlotCallbackRegistry.getEntryCallbacks()) &#123;</span><br><span class="line">            		handler.onPass(context, resourceWrapper, node, count, args);</span><br><span class="line">        	&#125;</span><br><span class="line">    	&#125; <span class="keyword">catch</span> (BlockException e) &#123;     <span class="comment">// @7                                                                                                              </span></span><br><span class="line">        	<span class="comment">// Blocked, set block exception to current entry.</span></span><br><span class="line">        	context.getCurEntry().setError(e);</span><br><span class="line">		<span class="comment">// Add block count.</span></span><br><span class="line">        	node.increaseBlockQps(count);</span><br><span class="line">        	<span class="keyword">if</span> (context.getCurEntry().getOriginNode() != <span class="keyword">null</span>) &#123;</span><br><span class="line">            		context.getCurEntry().getOriginNode().increaseBlockQps(count);</span><br><span class="line">        	&#125;</span><br><span class="line">		<span class="keyword">if</span> (resourceWrapper.getEntryType() == EntryType.IN) &#123;</span><br><span class="line">            		<span class="comment">// Add count for global inbound entry node for global statistics.</span></span><br><span class="line">            		Constants.ENTRY_NODE.increaseBlockQps(count);</span><br><span class="line">        	&#125;</span><br><span class="line">        	<span class="comment">// Handle block event with registered entry callback handlers.</span></span><br><span class="line">        	<span class="keyword">for</span> (ProcessorSlotEntryCallback&lt;DefaultNode&gt; handler : StatisticSlotCallbackRegistry.getEntryCallbacks()) &#123;</span><br><span class="line">            		handler.onBlocked(e, context, resourceWrapper, node, count, args);</span><br><span class="line">        	&#125;</span><br><span class="line">		<span class="keyword">throw</span> e;</span><br><span class="line">    	&#125; <span class="keyword">catch</span> (Throwable e) &#123;   <span class="comment">// @8</span></span><br><span class="line">        	<span class="comment">// Unexpected error, set error to current entry.</span></span><br><span class="line">        	context.getCurEntry().setError(e);</span><br><span class="line">		<span class="comment">// This should not happen.</span></span><br><span class="line">        	node.increaseExceptionQps(count);</span><br><span class="line">        	<span class="keyword">if</span> (context.getCurEntry().getOriginNode() != <span class="keyword">null</span>) &#123;</span><br><span class="line">            		context.getCurEntry().getOriginNode().increaseExceptionQps(count);</span><br><span class="line">        	&#125;</span><br><span class="line">		<span class="keyword">if</span> (resourceWrapper.getEntryType() == EntryType.IN) &#123;</span><br><span class="line">            		Constants.ENTRY_NODE.increaseExceptionQps(count);</span><br><span class="line">        	&#125;</span><br><span class="line">        	<span class="keyword">throw</span> e;</span><br><span class="line">    	&#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>代码@1：首先调用 fireEntry，先调用 Sentinel Slot Chain 中其他的处理器，执行完其他处理器的逻辑，例如 FlowSlot、DegradeSlot，因为 StatisticSlot 的职责是收集统计信息。</p>
<p>代码@2：如果后续处理器成功执行，则将正在执行线程数统计指标加一，并将通过的请求数量指标增加对应的值。下文会对 Sentinel Node 体系进行详细的介绍，在 Sentinel 中使用 Node 来表示调用链中的某一个节点，每个节点关联一个资源，资源的实时统计信息就存储在 Node 中，故该部分也是调用 DefaultNode 的相关方法来改变线程数等，将在下文会向详细介绍。</p>
<p>代码@3：如果上下文环境中保存了调用的源头（调用方）的节点信息不为空，则更新该节点的统计数据：线程数与通过数量。</p>
<p>代码@4：如果资源的进入类型为 EntryType.IN，表示入站流量，更新入站全局统计数据(集群范围 ClusterNode)。</p>
<p>代码@5：执行注册的进入Handler，可以通过 StatisticSlotCallbackRegistry 的 addEntryCallback 注册相关监听器。</p>
<p>代码@6：如果捕获到 PriorityWaitException ，则认为是等待过一定时间，但最终还是算通过，只需增加线程的个数，但无需增加节点通过的数量，具体原因我们在详细分析限流部分时会重点讨论，也会再次阐述 PriorityWaitException 的含义。</p>
<p>代码@7：如果捕获到 BlockException，则主要增加阻塞的数量。</p>
<p>代码@8：如果是系统异常，则增加异常数量。</p>
<p>我想上面的代码应该不难理解，但涉及到统计指标数据的变化，都是调用 DefaultNode node 相关的方法，从这里也可以看出，Node 将是实时统计数据的直接持有者，那毋容置疑接下来将重点来学习 Node，为了知识体系的完备性，我们先来看一下 StatisticSlot 的 exit 方法。</p>
          <!--noindex-->
          <div class="post-button text-center">
            <a class="btn" href="/posts/5f8f7185.html#more" rel="contents">
              阅读全文 &raquo;
            </a>
          </div>
          <!--/noindex-->
        
      
    </div>
    
    
    

    

    

    

    <footer class="post-footer">
      

      

      

      
      
        <div class="post-eof"></div>
      
    </footer>
  </div>
  
  
  
  </article>


    
  </section>

  
  <nav class="pagination">
    <a class="extend prev" rel="prev" href="/page/10/">&lt;</a><a class="page-number" href="/">1</a><span class="space">&hellip;</span><a class="page-number" href="/page/10/">10</a><span class="page-number current">11</span><a class="page-number" href="/page/12/">12</a><span class="space">&hellip;</span><a class="page-number" href="/page/14/">14</a><a class="extend next" rel="next" href="/page/12/">&gt;</a>
  </nav>



          </div>
          


          

        </div>
        
          
  
  <div class="sidebar-toggle">
    <div class="sidebar-toggle-line-wrap">
      <span class="sidebar-toggle-line sidebar-toggle-line-first"></span>
      <span class="sidebar-toggle-line sidebar-toggle-line-middle"></span>
      <span class="sidebar-toggle-line sidebar-toggle-line-last"></span>
    </div>
  </div>

  <aside id="sidebar" class="sidebar">
    
    <div class="sidebar-inner">

      

      

      <section class="site-overview-wrap sidebar-panel sidebar-panel-active">
        <div class="site-overview">
          <div class="site-author motion-element" itemprop="author" itemscope itemtype="http://schema.org/Person">
            
              <p class="site-author-name" itemprop="name"></p>
              <p class="site-description motion-element" itemprop="description"></p>
          </div>

          <nav class="site-state motion-element">

            
              <div class="site-state-item site-state-posts">
              
                <a href="/archives/">
              
                  <span class="site-state-item-count">139</span>
                  <span class="site-state-item-name">日志</span>
                </a>
              </div>
            

            
              
              
              <div class="site-state-item site-state-categories">
                <a href="/categories/index.html">
                  <span class="site-state-item-count">18</span>
                  <span class="site-state-item-name">分类</span>
                </a>
              </div>
            

            

          </nav>

          

          

          
          

          
          

          

        </div>
      </section>

      

      

    </div>
  </aside>


        
      </div>
    </main>

    <footer id="footer" class="footer">
      <div class="footer-inner">
        <div class="copyright">&copy; <span itemprop="copyrightYear">2021</span>
  <span class="with-love">
    <i class="fa fa-user"></i>
  </span>
  <span class="author" itemprop="copyrightHolder">中间件兴趣圈</span>

  
</div>


  <div class="powered-by">由 <a class="theme-link" target="_blank" href="https://hexo.io">Hexo</a> 强力驱动</div>



  <span class="post-meta-divider">|</span>



  <div class="theme-info">主题 &mdash; <a class="theme-link" target="_blank" href="https://github.com/iissnan/hexo-theme-next">NexT.Muse</a> v5.1.4</div>




        
<div class="busuanzi-count">
  <script async src="https://dn-lbstatics.qbox.me/busuanzi/2.3/busuanzi.pure.mini.js"></script>

  
    <span class="site-uv">
      <i class="fa fa-user"></i>
      <span class="busuanzi-value" id="busuanzi_value_site_uv"></span>
      
    </span>
  

  
    <span class="site-pv">
      <i class="fa fa-eye"></i>
      <span class="busuanzi-value" id="busuanzi_value_site_pv"></span>
      
    </span>
  
</div>








        
      </div>
    </footer>

    
      <div class="back-to-top">
        <i class="fa fa-arrow-up"></i>
        
      </div>
    

    

  </div>

  

<script type="text/javascript">
  if (Object.prototype.toString.call(window.Promise) !== '[object Function]') {
    window.Promise = null;
  }
</script>









  












  
  
    <script type="text/javascript" src="/lib/jquery/index.js?v=2.1.3"></script>
  

  
  
    <script type="text/javascript" src="/lib/fastclick/lib/fastclick.min.js?v=1.0.6"></script>
  

  
  
    <script type="text/javascript" src="/lib/jquery_lazyload/jquery.lazyload.js?v=1.9.7"></script>
  

  
  
    <script type="text/javascript" src="/lib/velocity/velocity.min.js?v=1.2.1"></script>
  

  
  
    <script type="text/javascript" src="/lib/velocity/velocity.ui.min.js?v=1.2.1"></script>
  

  
  
    <script type="text/javascript" src="/lib/fancybox/source/jquery.fancybox.pack.js?v=2.1.5"></script>
  


  


  <script type="text/javascript" src="/js/src/utils.js?v=5.1.4"></script>

  <script type="text/javascript" src="/js/src/motion.js?v=5.1.4"></script>



  
  

  

  


  <script type="text/javascript" src="/js/src/bootstrap.js?v=5.1.4"></script>



  


  




	





  





  












  





  

  
  <script src="https://cdn1.lncld.net/static/js/av-core-mini-0.6.4.js"></script>
  <script>AV.initialize("NNEhOL0iOcflg8f1U3HUqiCq-gzGzoHsz", "7kSmkbbb3DktmHALlShDsBUF");</script>
  <script>
    function showTime(Counter) {
      var query = new AV.Query(Counter);
      var entries = [];
      var $visitors = $(".leancloud_visitors");

      $visitors.each(function () {
        entries.push( $(this).attr("id").trim() );
      });

      query.containedIn('url', entries);
      query.find()
        .done(function (results) {
          var COUNT_CONTAINER_REF = '.leancloud-visitors-count';

          if (results.length === 0) {
            $visitors.find(COUNT_CONTAINER_REF).text(0);
            return;
          }

          for (var i = 0; i < results.length; i++) {
            var item = results[i];
            var url = item.get('url');
            var time = item.get('time');
            var element = document.getElementById(url);

            $(element).find(COUNT_CONTAINER_REF).text(time);
          }
          for(var i = 0; i < entries.length; i++) {
            var url = entries[i];
            var element = document.getElementById(url);
            var countSpan = $(element).find(COUNT_CONTAINER_REF);
            if( countSpan.text() == '') {
              countSpan.text(0);
            }
          }
        })
        .fail(function (object, error) {
          console.log("Error: " + error.code + " " + error.message);
        });
    }

    function addCount(Counter) {
      var $visitors = $(".leancloud_visitors");
      var url = $visitors.attr('id').trim();
      var title = $visitors.attr('data-flag-title').trim();
      var query = new AV.Query(Counter);

      query.equalTo("url", url);
      query.find({
        success: function(results) {
          if (results.length > 0) {
            var counter = results[0];
            counter.fetchWhenSave(true);
            counter.increment("time");
            counter.save(null, {
              success: function(counter) {
                var $element = $(document.getElementById(url));
                $element.find('.leancloud-visitors-count').text(counter.get('time'));
              },
              error: function(counter, error) {
                console.log('Failed to save Visitor num, with error message: ' + error.message);
              }
            });
          } else {
            var newcounter = new Counter();
            /* Set ACL */
            var acl = new AV.ACL();
            acl.setPublicReadAccess(true);
            acl.setPublicWriteAccess(true);
            newcounter.setACL(acl);
            /* End Set ACL */
            newcounter.set("title", title);
            newcounter.set("url", url);
            newcounter.set("time", 1);
            newcounter.save(null, {
              success: function(newcounter) {
                var $element = $(document.getElementById(url));
                $element.find('.leancloud-visitors-count').text(newcounter.get('time'));
              },
              error: function(newcounter, error) {
                console.log('Failed to create');
              }
            });
          }
        },
        error: function(error) {
          console.log('Error:' + error.code + " " + error.message);
        }
      });
    }

    $(function() {
      var Counter = AV.Object.extend("Counter");
      if ($('.leancloud_visitors').length == 1) {
        addCount(Counter);
      } else if ($('.post-title-link').length > 1) {
        showTime(Counter);
      }
    });
  </script>



  

  

  
  

  

  

  

</body>
</html>
